Fix eviction when random_pos selects the end

- Do not try to allocate more DB blocks in an inode block until it's "confirmed" and "locked" by the first write
- Do not recheck for new zero DB blocks on first write into an inode block - a CAS failure means someone else is already writing into it
- Throw new allocation blocks away regardless of whether the known_version is 0 on a CAS failure

docs/config/src/network.yml

@@ -245,8 +245,8 @@
     повторная попытка соединения.
 - name: up_wait_retry_interval
   type: ms
-  min: 50
-  default: 500
+  min: 10
+  default: 50
   online: true
   info: |
     OSDs respond to clients with a special error code when they receive I/O

mon/mon.js

@@ -92,7 +92,7 @@ const etcd_tree = {
             peer_connect_timeout: 5, // seconds. min: 1
             osd_idle_timeout: 5, // seconds. min: 1
             osd_ping_timeout: 5, // seconds. min: 1
-            up_wait_retry_interval: 500, // ms. min: 50
+            up_wait_retry_interval: 50, // ms. min: 10
             max_etcd_attempts: 5,
             etcd_quick_timeout: 1000, // ms
             etcd_slow_timeout: 5000, // ms

patches/libvirt-9.10-vitastor.diff

@@ -0,0 +1,643 @@
+commit c1cd026e211e94b120028e7c98a6e4ce5afe9846
+Author: Vitaliy Filippov <vitalif@yourcmc.ru>
+Date:   Wed Jan 24 22:04:50 2024 +0300
+
+    Add Vitastor support
+
+diff --git a/include/libvirt/libvirt-storage.h b/include/libvirt/libvirt-storage.h
+index aaad4a3da1..5f5daa8341 100644
+--- a/include/libvirt/libvirt-storage.h
++++ b/include/libvirt/libvirt-storage.h
+@@ -326,6 +326,7 @@ typedef enum {
+     VIR_CONNECT_LIST_STORAGE_POOLS_ZFS           = 1 << 17, /* (Since: 1.2.8) */
+     VIR_CONNECT_LIST_STORAGE_POOLS_VSTORAGE      = 1 << 18, /* (Since: 3.1.0) */
+     VIR_CONNECT_LIST_STORAGE_POOLS_ISCSI_DIRECT  = 1 << 19, /* (Since: 5.6.0) */
++    VIR_CONNECT_LIST_STORAGE_POOLS_VITASTOR      = 1 << 20, /* (Since: 5.0.0) */
+ } virConnectListAllStoragePoolsFlags;
+ 
+ int                     virConnectListAllStoragePools(virConnectPtr conn,
+diff --git a/src/conf/domain_conf.c b/src/conf/domain_conf.c
+index 22ad43e1d7..56c81d6852 100644
+--- a/src/conf/domain_conf.c
++++ b/src/conf/domain_conf.c
+@@ -7185,7 +7185,8 @@ virDomainDiskSourceNetworkParse(xmlNodePtr node,
+     src->configFile = virXPathString("string(./config/@file)", ctxt);
+ 
+     if (src->protocol == VIR_STORAGE_NET_PROTOCOL_HTTP ||
+-        src->protocol == VIR_STORAGE_NET_PROTOCOL_HTTPS)
++        src->protocol == VIR_STORAGE_NET_PROTOCOL_HTTPS ||
++        src->protocol == VIR_STORAGE_NET_PROTOCOL_VITASTOR)
+         src->query = virXMLPropString(node, "query");
+ 
+     if (virDomainStorageNetworkParseHosts(node, ctxt, &src->hosts, &src->nhosts) < 0)
+@@ -30618,6 +30619,7 @@ virDomainStorageSourceTranslateSourcePool(virStorageSource *src,
+ 
+     case VIR_STORAGE_POOL_MPATH:
+     case VIR_STORAGE_POOL_RBD:
++    case VIR_STORAGE_POOL_VITASTOR:
+     case VIR_STORAGE_POOL_SHEEPDOG:
+     case VIR_STORAGE_POOL_GLUSTER:
+     case VIR_STORAGE_POOL_LAST:
+diff --git a/src/conf/domain_validate.c b/src/conf/domain_validate.c
+index c72108886e..c739ed6c43 100644
+--- a/src/conf/domain_validate.c
++++ b/src/conf/domain_validate.c
+@@ -495,6 +495,7 @@ virDomainDiskDefValidateSourceChainOne(const virStorageSource *src)
+         case VIR_STORAGE_NET_PROTOCOL_RBD:
+             break;
+ 
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+         case VIR_STORAGE_NET_PROTOCOL_NBD:
+         case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+         case VIR_STORAGE_NET_PROTOCOL_GLUSTER:
+@@ -541,7 +542,7 @@ virDomainDiskDefValidateSourceChainOne(const virStorageSource *src)
+         }
+     }
+ 
+-    /* internal snapshots and config files are currently supported only with rbd: */
++    /* internal snapshots are currently supported only with rbd: */
+     if (virStorageSourceGetActualType(src) != VIR_STORAGE_TYPE_NETWORK &&
+         src->protocol != VIR_STORAGE_NET_PROTOCOL_RBD) {
+         if (src->snapshot) {
+@@ -549,10 +550,15 @@ virDomainDiskDefValidateSourceChainOne(const virStorageSource *src)
+                            _("<snapshot> element is currently supported only with 'rbd' disks"));
+             return -1;
+         }
++    }
+ 
++    /* config files are currently supported only with rbd and vitastor: */
++    if (virStorageSourceGetActualType(src) != VIR_STORAGE_TYPE_NETWORK &&
++        src->protocol != VIR_STORAGE_NET_PROTOCOL_RBD &&
++        src->protocol != VIR_STORAGE_NET_PROTOCOL_VITASTOR) {
+         if (src->configFile) {
+             virReportError(VIR_ERR_XML_ERROR, "%s",
+-                           _("<config> element is currently supported only with 'rbd' disks"));
++                           _("<config> element is currently supported only with 'rbd' and 'vitastor' disks"));
+             return -1;
+         }
+     }
+diff --git a/src/conf/schemas/domaincommon.rng b/src/conf/schemas/domaincommon.rng
+index b98a2ae602..7d7a872e01 100644
+--- a/src/conf/schemas/domaincommon.rng
++++ b/src/conf/schemas/domaincommon.rng
+@@ -1997,6 +1997,35 @@
+     </element>
+   </define>
+ 
++  <define name="diskSourceNetworkProtocolVitastor">
++    <element name="source">
++      <interleave>
++        <attribute name="protocol">
++          <value>vitastor</value>
++        </attribute>
++        <ref name="diskSourceCommon"/>
++        <optional>
++          <attribute name="name"/>
++        </optional>
++        <optional>
++          <attribute name="query"/>
++        </optional>
++        <zeroOrMore>
++          <ref name="diskSourceNetworkHost"/>
++        </zeroOrMore>
++        <optional>
++          <element name="config">
++            <attribute name="file">
++              <ref name="absFilePath"/>
++            </attribute>
++            <empty/>
++          </element>
++        </optional>
++        <empty/>
++      </interleave>
++    </element>
++  </define>
++
+   <define name="diskSourceNetworkProtocolISCSI">
+     <element name="source">
+       <attribute name="protocol">
+@@ -2347,6 +2376,7 @@
+       <ref name="diskSourceNetworkProtocolSimple"/>
+       <ref name="diskSourceNetworkProtocolVxHS"/>
+       <ref name="diskSourceNetworkProtocolNFS"/>
++      <ref name="diskSourceNetworkProtocolVitastor"/>
+     </choice>
+   </define>
+ 
+diff --git a/src/conf/storage_conf.c b/src/conf/storage_conf.c
+index 68842004b7..1d69a788b6 100644
+--- a/src/conf/storage_conf.c
++++ b/src/conf/storage_conf.c
+@@ -56,7 +56,7 @@ VIR_ENUM_IMPL(virStoragePool,
+               "logical", "disk", "iscsi",
+               "iscsi-direct", "scsi", "mpath",
+               "rbd", "sheepdog", "gluster",
+-              "zfs", "vstorage",
++              "zfs", "vstorage", "vitastor",
+ );
+ 
+ VIR_ENUM_IMPL(virStoragePoolFormatFileSystem,
+@@ -242,6 +242,18 @@ static virStoragePoolTypeInfo poolTypeInfo[] = {
+           .formatToString = virStorageFileFormatTypeToString,
+       }
+     },
++    {.poolType = VIR_STORAGE_POOL_VITASTOR,
++     .poolOptions = {
++         .flags = (VIR_STORAGE_POOL_SOURCE_HOST |
++                   VIR_STORAGE_POOL_SOURCE_NETWORK |
++                   VIR_STORAGE_POOL_SOURCE_NAME),
++      },
++      .volOptions = {
++          .defaultFormat = VIR_STORAGE_FILE_RAW,
++          .formatFromString = virStorageVolumeFormatFromString,
++          .formatToString = virStorageFileFormatTypeToString,
++      }
++    },
+     {.poolType = VIR_STORAGE_POOL_SHEEPDOG,
+      .poolOptions = {
+          .flags = (VIR_STORAGE_POOL_SOURCE_HOST |
+@@ -538,6 +550,11 @@ virStoragePoolDefParseSource(xmlXPathContextPtr ctxt,
+                        _("element 'name' is mandatory for RBD pool"));
+         return -1;
+     }
++    if (pool_type == VIR_STORAGE_POOL_VITASTOR && source->name == NULL) {
++        virReportError(VIR_ERR_XML_ERROR, "%s",
++                       _("element 'name' is mandatory for Vitastor pool"));
++        return -1;
++    }
+ 
+     if (options->formatFromString) {
+         g_autofree char *format = NULL;
+@@ -1127,6 +1144,7 @@ virStoragePoolDefFormatBuf(virBuffer *buf,
+     /* RBD, Sheepdog, Gluster and Iscsi-direct devices are not local block devs nor
+      * files, so they don't have a target */
+     if (def->type != VIR_STORAGE_POOL_RBD &&
++        def->type != VIR_STORAGE_POOL_VITASTOR &&
+         def->type != VIR_STORAGE_POOL_SHEEPDOG &&
+         def->type != VIR_STORAGE_POOL_GLUSTER &&
+         def->type != VIR_STORAGE_POOL_ISCSI_DIRECT) {
+diff --git a/src/conf/storage_conf.h b/src/conf/storage_conf.h
+index fc67957cfe..720c07ef74 100644
+--- a/src/conf/storage_conf.h
++++ b/src/conf/storage_conf.h
+@@ -103,6 +103,7 @@ typedef enum {
+     VIR_STORAGE_POOL_GLUSTER,  /* Gluster device */
+     VIR_STORAGE_POOL_ZFS,      /* ZFS */
+     VIR_STORAGE_POOL_VSTORAGE, /* Virtuozzo Storage */
++    VIR_STORAGE_POOL_VITASTOR, /* Vitastor */
+ 
+     VIR_STORAGE_POOL_LAST,
+ } virStoragePoolType;
+@@ -454,6 +455,7 @@ VIR_ENUM_DECL(virStoragePartedFs);
+                  VIR_CONNECT_LIST_STORAGE_POOLS_SCSI     | \
+                  VIR_CONNECT_LIST_STORAGE_POOLS_MPATH    | \
+                  VIR_CONNECT_LIST_STORAGE_POOLS_RBD      | \
++                 VIR_CONNECT_LIST_STORAGE_POOLS_VITASTOR | \
+                  VIR_CONNECT_LIST_STORAGE_POOLS_SHEEPDOG | \
+                  VIR_CONNECT_LIST_STORAGE_POOLS_GLUSTER  | \
+                  VIR_CONNECT_LIST_STORAGE_POOLS_ZFS      | \
+diff --git a/src/conf/storage_source_conf.c b/src/conf/storage_source_conf.c
+index f974a521b1..cd394d0a9f 100644
+--- a/src/conf/storage_source_conf.c
++++ b/src/conf/storage_source_conf.c
+@@ -88,6 +88,7 @@ VIR_ENUM_IMPL(virStorageNetProtocol,
+               "ssh",
+               "vxhs",
+               "nfs",
++              "vitastor",
+ );
+ 
+ 
+@@ -1301,6 +1302,7 @@ virStorageSourceNetworkDefaultPort(virStorageNetProtocol protocol)
+         case VIR_STORAGE_NET_PROTOCOL_GLUSTER:
+             return 24007;
+ 
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+         case VIR_STORAGE_NET_PROTOCOL_RBD:
+             /* we don't provide a default for RBD */
+             return 0;
+diff --git a/src/conf/storage_source_conf.h b/src/conf/storage_source_conf.h
+index 5e7d127453..283709eeb3 100644
+--- a/src/conf/storage_source_conf.h
++++ b/src/conf/storage_source_conf.h
+@@ -129,6 +129,7 @@ typedef enum {
+     VIR_STORAGE_NET_PROTOCOL_SSH,
+     VIR_STORAGE_NET_PROTOCOL_VXHS,
+     VIR_STORAGE_NET_PROTOCOL_NFS,
++    VIR_STORAGE_NET_PROTOCOL_VITASTOR,
+ 
+     VIR_STORAGE_NET_PROTOCOL_LAST
+ } virStorageNetProtocol;
+diff --git a/src/conf/virstorageobj.c b/src/conf/virstorageobj.c
+index 59fa5da372..4739167f5f 100644
+--- a/src/conf/virstorageobj.c
++++ b/src/conf/virstorageobj.c
+@@ -1438,6 +1438,7 @@ virStoragePoolObjSourceFindDuplicateCb(const void *payload,
+             return 1;
+         break;
+ 
++    case VIR_STORAGE_POOL_VITASTOR:
+     case VIR_STORAGE_POOL_ISCSI_DIRECT:
+     case VIR_STORAGE_POOL_RBD:
+     case VIR_STORAGE_POOL_LAST:
+@@ -1921,6 +1922,8 @@ virStoragePoolObjMatch(virStoragePoolObj *obj,
+                (obj->def->type == VIR_STORAGE_POOL_MPATH))   ||
+               (MATCH(VIR_CONNECT_LIST_STORAGE_POOLS_RBD) &&
+                (obj->def->type == VIR_STORAGE_POOL_RBD))     ||
++              (MATCH(VIR_CONNECT_LIST_STORAGE_POOLS_VITASTOR) &&
++               (obj->def->type == VIR_STORAGE_POOL_VITASTOR)) ||
+               (MATCH(VIR_CONNECT_LIST_STORAGE_POOLS_SHEEPDOG) &&
+                (obj->def->type == VIR_STORAGE_POOL_SHEEPDOG)) ||
+               (MATCH(VIR_CONNECT_LIST_STORAGE_POOLS_GLUSTER) &&
+diff --git a/src/libvirt-storage.c b/src/libvirt-storage.c
+index db7660aac4..561df34709 100644
+--- a/src/libvirt-storage.c
++++ b/src/libvirt-storage.c
+@@ -94,6 +94,7 @@ virStoragePoolGetConnect(virStoragePoolPtr pool)
+  * VIR_CONNECT_LIST_STORAGE_POOLS_SCSI
+  * VIR_CONNECT_LIST_STORAGE_POOLS_MPATH
+  * VIR_CONNECT_LIST_STORAGE_POOLS_RBD
++ * VIR_CONNECT_LIST_STORAGE_POOLS_VITASTOR
+  * VIR_CONNECT_LIST_STORAGE_POOLS_SHEEPDOG
+  * VIR_CONNECT_LIST_STORAGE_POOLS_GLUSTER
+  * VIR_CONNECT_LIST_STORAGE_POOLS_ZFS
+diff --git a/src/libxl/libxl_conf.c b/src/libxl/libxl_conf.c
+index 62e1be6672..71a1d42896 100644
+--- a/src/libxl/libxl_conf.c
++++ b/src/libxl/libxl_conf.c
+@@ -979,6 +979,7 @@ libxlMakeNetworkDiskSrcStr(virStorageSource *src,
+     case VIR_STORAGE_NET_PROTOCOL_SSH:
+     case VIR_STORAGE_NET_PROTOCOL_VXHS:
+     case VIR_STORAGE_NET_PROTOCOL_NFS:
++    case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+     case VIR_STORAGE_NET_PROTOCOL_LAST:
+     case VIR_STORAGE_NET_PROTOCOL_NONE:
+         virReportError(VIR_ERR_NO_SUPPORT,
+diff --git a/src/libxl/xen_xl.c b/src/libxl/xen_xl.c
+index f175359307..8efcf4c329 100644
+--- a/src/libxl/xen_xl.c
++++ b/src/libxl/xen_xl.c
+@@ -1456,6 +1456,7 @@ xenFormatXLDiskSrcNet(virStorageSource *src)
+     case VIR_STORAGE_NET_PROTOCOL_SSH:
+     case VIR_STORAGE_NET_PROTOCOL_VXHS:
+     case VIR_STORAGE_NET_PROTOCOL_NFS:
++    case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+     case VIR_STORAGE_NET_PROTOCOL_LAST:
+     case VIR_STORAGE_NET_PROTOCOL_NONE:
+         virReportError(VIR_ERR_NO_SUPPORT,
+diff --git a/src/qemu/qemu_block.c b/src/qemu/qemu_block.c
+index 7e9daf0bdc..825b4a3006 100644
+--- a/src/qemu/qemu_block.c
++++ b/src/qemu/qemu_block.c
+@@ -758,6 +758,38 @@ qemuBlockStorageSourceGetRBDProps(virStorageSource *src,
+ }
+ 
+ 
++static virJSONValue *
++qemuBlockStorageSourceGetVitastorProps(virStorageSource *src)
++{
++    virJSONValue *ret = NULL;
++    virStorageNetHostDef *host;
++    size_t i;
++    g_auto(virBuffer) buf = VIR_BUFFER_INITIALIZER;
++    g_autofree char *etcd = NULL;
++
++    for (i = 0; i < src->nhosts; i++) {
++        host = src->hosts + i;
++        if ((virStorageNetHostTransport)host->transport != VIR_STORAGE_NET_HOST_TRANS_TCP) {
++            return NULL;
++        }
++        virBufferAsprintf(&buf, i > 0 ? ",%s:%u" : "%s:%u", host->name, host->port);
++    }
++    if (src->nhosts > 0) {
++        etcd = virBufferContentAndReset(&buf);
++    }
++
++    if (virJSONValueObjectAdd(&ret,
++                              "S:etcd-host", etcd,
++                              "S:etcd-prefix", src->query,
++                              "S:config-path", src->configFile,
++                              "s:image", src->path,
++                              NULL) < 0)
++        return NULL;
++
++    return ret;
++}
++
++
+ static virJSONValue *
+ qemuBlockStorageSourceGetSheepdogProps(virStorageSource *src)
+ {
+@@ -1140,6 +1172,12 @@ qemuBlockStorageSourceGetBackendProps(virStorageSource *src,
+                 return NULL;
+             break;
+ 
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
++            driver = "vitastor";
++            if (!(fileprops = qemuBlockStorageSourceGetVitastorProps(src)))
++                return NULL;
++            break;
++
+         case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+             driver = "sheepdog";
+             if (!(fileprops = qemuBlockStorageSourceGetSheepdogProps(src)))
+@@ -2032,6 +2070,7 @@ qemuBlockGetBackingStoreString(virStorageSource *src,
+ 
+             case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+             case VIR_STORAGE_NET_PROTOCOL_RBD:
++            case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+             case VIR_STORAGE_NET_PROTOCOL_VXHS:
+             case VIR_STORAGE_NET_PROTOCOL_NFS:
+             case VIR_STORAGE_NET_PROTOCOL_SSH:
+@@ -2415,6 +2454,12 @@ qemuBlockStorageSourceCreateGetStorageProps(virStorageSource *src,
+                 return -1;
+             break;
+ 
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
++            driver = "vitastor";
++            if (!(location = qemuBlockStorageSourceGetVitastorProps(src)))
++                return -1;
++            break;
++
+         case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+             driver = "sheepdog";
+             if (!(location = qemuBlockStorageSourceGetSheepdogProps(src)))
+diff --git a/src/qemu/qemu_domain.c b/src/qemu/qemu_domain.c
+index 953808fcfe..62860283d8 100644
+--- a/src/qemu/qemu_domain.c
++++ b/src/qemu/qemu_domain.c
+@@ -5215,7 +5215,8 @@ qemuDomainValidateStorageSource(virStorageSource *src,
+     if (src->query &&
+         (actualType != VIR_STORAGE_TYPE_NETWORK ||
+          (src->protocol != VIR_STORAGE_NET_PROTOCOL_HTTPS &&
+-          src->protocol != VIR_STORAGE_NET_PROTOCOL_HTTP))) {
++          src->protocol != VIR_STORAGE_NET_PROTOCOL_HTTP &&
++          src->protocol != VIR_STORAGE_NET_PROTOCOL_VITASTOR))) {
+         virReportError(VIR_ERR_CONFIG_UNSUPPORTED, "%s",
+                        _("query is supported only with HTTP(S) protocols"));
+         return -1;
+@@ -10340,6 +10341,7 @@ qemuDomainPrepareStorageSourceTLS(virStorageSource *src,
+         break;
+ 
+     case VIR_STORAGE_NET_PROTOCOL_RBD:
++    case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+     case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+     case VIR_STORAGE_NET_PROTOCOL_GLUSTER:
+     case VIR_STORAGE_NET_PROTOCOL_ISCSI:
+diff --git a/src/qemu/qemu_snapshot.c b/src/qemu/qemu_snapshot.c
+index 73ff533827..e9c799ca8f 100644
+--- a/src/qemu/qemu_snapshot.c
++++ b/src/qemu/qemu_snapshot.c
+@@ -423,6 +423,7 @@ qemuSnapshotPrepareDiskExternalInactive(virDomainSnapshotDiskDef *snapdisk,
+         case VIR_STORAGE_NET_PROTOCOL_NONE:
+         case VIR_STORAGE_NET_PROTOCOL_NBD:
+         case VIR_STORAGE_NET_PROTOCOL_RBD:
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+         case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+         case VIR_STORAGE_NET_PROTOCOL_GLUSTER:
+         case VIR_STORAGE_NET_PROTOCOL_ISCSI:
+@@ -648,6 +649,7 @@ qemuSnapshotPrepareDiskInternal(virDomainDiskDef *disk,
+         case VIR_STORAGE_NET_PROTOCOL_NONE:
+         case VIR_STORAGE_NET_PROTOCOL_NBD:
+         case VIR_STORAGE_NET_PROTOCOL_RBD:
++        case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
+         case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+         case VIR_STORAGE_NET_PROTOCOL_GLUSTER:
+         case VIR_STORAGE_NET_PROTOCOL_ISCSI:
+diff --git a/src/storage/storage_driver.c b/src/storage/storage_driver.c
+index 314fe930e0..fb615a8b4e 100644
+--- a/src/storage/storage_driver.c
++++ b/src/storage/storage_driver.c
+@@ -1626,6 +1626,7 @@ storageVolLookupByPathCallback(virStoragePoolObj *obj,
+ 
+         case VIR_STORAGE_POOL_GLUSTER:
+         case VIR_STORAGE_POOL_RBD:
++        case VIR_STORAGE_POOL_VITASTOR:
+         case VIR_STORAGE_POOL_SHEEPDOG:
+         case VIR_STORAGE_POOL_ZFS:
+         case VIR_STORAGE_POOL_LAST:
+diff --git a/src/storage_file/storage_source_backingstore.c b/src/storage_file/storage_source_backingstore.c
+index 80681924ea..8a3ade9ec0 100644
+--- a/src/storage_file/storage_source_backingstore.c
++++ b/src/storage_file/storage_source_backingstore.c
+@@ -287,6 +287,75 @@ virStorageSourceParseRBDColonString(const char *rbdstr,
+ }
+ 
+ 
++static int
++virStorageSourceParseVitastorColonString(const char *colonstr,
++                                         virStorageSource *src)
++{
++    char *p, *e, *next;
++    g_autofree char *options = NULL;
++
++    /* optionally skip the "vitastor:" prefix if provided */
++    if (STRPREFIX(colonstr, "vitastor:"))
++        colonstr += strlen("vitastor:");
++
++    options = g_strdup(colonstr);
++
++    p = options;
++    while (*p) {
++        /* find : delimiter or end of string */
++        for (e = p; *e && *e != ':'; ++e) {
++            if (*e == '\\') {
++                e++;
++                if (*e == '\0')
++                    break;
++            }
++        }
++        if (*e == '\0') {
++            next = e;    /* last kv pair */
++        } else {
++            next = e + 1;
++            *e = '\0';
++        }
++
++        if (STRPREFIX(p, "image=")) {
++            src->path = g_strdup(p + strlen("image="));
++        } else if (STRPREFIX(p, "etcd-prefix=")) {
++            src->query = g_strdup(p + strlen("etcd-prefix="));
++        } else if (STRPREFIX(p, "config-path=")) {
++            src->configFile = g_strdup(p + strlen("config-path="));
++        } else if (STRPREFIX(p, "etcd-host=")) {
++            char *h, *sep;
++
++            h = p + strlen("etcd-host=");
++            while (h < e) {
++                for (sep = h; sep < e; ++sep) {
++                    if (*sep == '\\' && (sep[1] == ',' ||
++                                         sep[1] == ';' ||
++                                         sep[1] == ' ')) {
++                        *sep = '\0';
++                        sep += 2;
++                        break;
++                    }
++                }
++
++                if (virStorageSourceRBDAddHost(src, h) < 0)
++                    return -1;
++
++                h = sep;
++            }
++        }
++
++        p = next;
++    }
++
++    if (!src->path) {
++        return -1;
++    }
++
++    return 0;
++}
++
++
+ static int
+ virStorageSourceParseNBDColonString(const char *nbdstr,
+                                     virStorageSource *src)
+@@ -399,6 +468,11 @@ virStorageSourceParseBackingColon(virStorageSource *src,
+             return -1;
+         break;
+ 
++    case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
++        if (virStorageSourceParseVitastorColonString(path, src) < 0)
++            return -1;
++        break;
++
+     case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
+     case VIR_STORAGE_NET_PROTOCOL_LAST:
+     case VIR_STORAGE_NET_PROTOCOL_NONE:
+@@ -975,6 +1049,54 @@ virStorageSourceParseBackingJSONRBD(virStorageSource *src,
+     return 0;
+ }
+ 
++static int
++virStorageSourceParseBackingJSONVitastor(virStorageSource *src,
++                                         virJSONValue *json,
++                                         const char *jsonstr G_GNUC_UNUSED,
++                                         int opaque G_GNUC_UNUSED)
++{
++    const char *filename;
++    const char *image = virJSONValueObjectGetString(json, "image");
++    const char *conf = virJSONValueObjectGetString(json, "config-path");
++    const char *etcd_prefix = virJSONValueObjectGetString(json, "etcd-prefix");
++    virJSONValue *servers = virJSONValueObjectGetArray(json, "server");
++    size_t nservers;
++    size_t i;
++
++    src->type = VIR_STORAGE_TYPE_NETWORK;
++    src->protocol = VIR_STORAGE_NET_PROTOCOL_VITASTOR;
++
++    /* legacy syntax passed via 'filename' option */
++    if ((filename = virJSONValueObjectGetString(json, "filename")))
++        return virStorageSourceParseVitastorColonString(filename, src);
++
++    if (!image) {
++        virReportError(VIR_ERR_INVALID_ARG, "%s",
++                       _("missing image name in Vitastor backing volume "
++                         "JSON specification"));
++        return -1;
++    }
++
++    src->path = g_strdup(image);
++    src->configFile = g_strdup(conf);
++    src->query = g_strdup(etcd_prefix);
++
++    if (servers) {
++        nservers = virJSONValueArraySize(servers);
++
++        src->hosts = g_new0(virStorageNetHostDef, nservers);
++        src->nhosts = nservers;
++
++        for (i = 0; i < nservers; i++) {
++            if (virStorageSourceParseBackingJSONInetSocketAddress(src->hosts + i,
++                                                                  virJSONValueArrayGet(servers, i)) < 0)
++                return -1;
++        }
++    }
++
++    return 0;
++}
++
+ static int
+ virStorageSourceParseBackingJSONRaw(virStorageSource *src,
+                                     virJSONValue *json,
+@@ -1152,6 +1274,7 @@ static const struct virStorageSourceJSONDriverParser jsonParsers[] = {
+     {"sheepdog", false, virStorageSourceParseBackingJSONSheepdog, 0},
+     {"ssh", false, virStorageSourceParseBackingJSONSSH, 0},
+     {"rbd", false, virStorageSourceParseBackingJSONRBD, 0},
++    {"vitastor", false, virStorageSourceParseBackingJSONVitastor, 0},
+     {"raw", true, virStorageSourceParseBackingJSONRaw, 0},
+     {"nfs", false, virStorageSourceParseBackingJSONNFS, 0},
+     {"vxhs", false, virStorageSourceParseBackingJSONVxHS, 0},
+diff --git a/src/test/test_driver.c b/src/test/test_driver.c
+index e87d7cfd44..ccc05d7aae 100644
+--- a/src/test/test_driver.c
++++ b/src/test/test_driver.c
+@@ -7335,6 +7335,7 @@ testStorageVolumeTypeForPool(int pooltype)
+     case VIR_STORAGE_POOL_ISCSI_DIRECT:
+     case VIR_STORAGE_POOL_GLUSTER:
+     case VIR_STORAGE_POOL_RBD:
++    case VIR_STORAGE_POOL_VITASTOR:
+         return VIR_STORAGE_VOL_NETWORK;
+     case VIR_STORAGE_POOL_LOGICAL:
+     case VIR_STORAGE_POOL_DISK:
+diff --git a/tests/storagepoolcapsschemadata/poolcaps-fs.xml b/tests/storagepoolcapsschemadata/poolcaps-fs.xml
+index eee75af746..8bd0a57bdd 100644
+--- a/tests/storagepoolcapsschemadata/poolcaps-fs.xml
++++ b/tests/storagepoolcapsschemadata/poolcaps-fs.xml
+@@ -204,4 +204,11 @@
+       </enum>
+     </volOptions>
+   </pool>
++  <pool type='vitastor' supported='no'>
++    <volOptions>
++      <defaultFormat type='raw'/>
++      <enum name='targetFormatType'>
++      </enum>
++    </volOptions>
++  </pool>
+ </storagepoolCapabilities>
+diff --git a/tests/storagepoolcapsschemadata/poolcaps-full.xml b/tests/storagepoolcapsschemadata/poolcaps-full.xml
+index 805950a937..852df0de16 100644
+--- a/tests/storagepoolcapsschemadata/poolcaps-full.xml
++++ b/tests/storagepoolcapsschemadata/poolcaps-full.xml
+@@ -204,4 +204,11 @@
+       </enum>
+     </volOptions>
+   </pool>
++  <pool type='vitastor' supported='yes'>
++    <volOptions>
++      <defaultFormat type='raw'/>
++      <enum name='targetFormatType'>
++      </enum>
++    </volOptions>
++  </pool>
+ </storagepoolCapabilities>
+diff --git a/tests/storagepoolxml2argvtest.c b/tests/storagepoolxml2argvtest.c
+index e8e40d695e..db55fe5f3a 100644
+--- a/tests/storagepoolxml2argvtest.c
++++ b/tests/storagepoolxml2argvtest.c
+@@ -65,6 +65,7 @@ testCompareXMLToArgvFiles(bool shouldFail,
+     case VIR_STORAGE_POOL_GLUSTER:
+     case VIR_STORAGE_POOL_ZFS:
+     case VIR_STORAGE_POOL_VSTORAGE:
++    case VIR_STORAGE_POOL_VITASTOR:
+     case VIR_STORAGE_POOL_LAST:
+     default:
+         VIR_TEST_DEBUG("pool type '%s' has no xml2argv test", defTypeStr);
+diff --git a/tools/virsh-pool.c b/tools/virsh-pool.c
+index 36f00cf643..5f5bd3464e 100644
+--- a/tools/virsh-pool.c
++++ b/tools/virsh-pool.c
+@@ -1223,6 +1223,9 @@ cmdPoolList(vshControl *ctl, const vshCmd *cmd G_GNUC_UNUSED)
+             case VIR_STORAGE_POOL_VSTORAGE:
+                 flags |= VIR_CONNECT_LIST_STORAGE_POOLS_VSTORAGE;
+                 break;
++            case VIR_STORAGE_POOL_VITASTOR:
++                flags |= VIR_CONNECT_LIST_STORAGE_POOLS_VITASTOR;
++                break;
+             case VIR_STORAGE_POOL_LAST:
+                 break;
+             }

src/CMakeLists.txt

@@ -181,6 +181,25 @@ target_link_libraries(vitastor-nbd
 	vitastor_client
 )
 
+# vitastor-kv
+add_executable(vitastor-kv
+	kv_cli.cpp
+	kv_db.cpp
+	kv_db.h
+)
+target_link_libraries(vitastor-kv
+	vitastor_client
+)
+
+add_executable(vitastor-kv-stress
+	kv_stress.cpp
+	kv_db.cpp
+	kv_db.h
+)
+target_link_libraries(vitastor-kv-stress
+	vitastor_client
+)
+
 # vitastor-nfs
 add_executable(vitastor-nfs
 	nfs_proxy.cpp

src/blockstore_impl.cpp

@@ -163,20 +163,10 @@ void blockstore_impl_t::loop()
             }
             else if (op->opcode == BS_OP_SYNC)
             {
-                // wait for all small writes to be submitted
-                // wait for all big writes to complete, submit data device fsync
+                // sync only completed writes?
                 // wait for the data device fsync to complete, then submit journal writes for big writes
                 // then submit an fsync operation
-                if (has_writes)
-                {
-                    // Can't submit SYNC before previous writes
-                    continue;
-                }
                 wr_st = continue_sync(op);
-                if (wr_st != 2)
-                {
-                    has_writes = wr_st > 0 ? 1 : 2;
-                }
             }
             else if (op->opcode == BS_OP_STABLE)
             {
@@ -283,7 +273,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_JOURNAL)
     {
-        if (journal.used_start == PRIV(op)->wait_detail)
+        if (journal.used_start == PRIV(op)->wait_detail && !unstable_count_changed)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG
@@ -291,6 +281,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
 #endif
             return;
         }
+        unstable_count_changed = false;
         flusher->release_trim();
         PRIV(op)->wait_for = 0;
     }
@@ -362,6 +353,7 @@ void blockstore_impl_t::enqueue_op(blockstore_op_t *op)
                     };
                 }
                 unstable_writes.clear();
+                unstable_count_changed = true;
                 op->callback = [old_callback](blockstore_op_t *op)
                 {
                     obj_ver_id *vers = (obj_ver_id*)op->buf;

src/blockstore_impl.h

@@ -55,6 +55,7 @@
 #define IS_JOURNAL(st) (((st) & 0x0F) == BS_ST_SMALL_WRITE)
 #define IS_BIG_WRITE(st) (((st) & 0x0F) == BS_ST_BIG_WRITE)
 #define IS_DELETE(st) (((st) & 0x0F) == BS_ST_DELETE)
+#define IS_INSTANT(st) (((st) & BS_ST_TYPE_MASK) == BS_ST_DELETE || ((st) & BS_ST_INSTANT))
 
 #define BS_SUBMIT_CHECK_SQES(n) \
     if (ringloop->sqes_left() < (n))\
@@ -275,6 +276,7 @@ class blockstore_impl_t
     std::vector<blockstore_op_t*> submit_queue;
     std::vector<obj_ver_id> unsynced_big_writes, unsynced_small_writes;
     int unsynced_big_write_count = 0, unstable_unsynced = 0;
+    bool unstable_count_changed = false;
     int unsynced_queued_ops = 0;
     allocator *data_alloc = NULL;
     uint64_t used_blocks = 0;
@@ -377,7 +379,7 @@ class blockstore_impl_t
     // Stabilize
     int dequeue_stable(blockstore_op_t *op);
     int continue_stable(blockstore_op_t *op);
-    void mark_stable(const obj_ver_id & ov, bool forget_dirty = false);
+    void mark_stable(obj_ver_id ov, bool forget_dirty = false);
     void stabilize_object(object_id oid, uint64_t max_ver);
     blockstore_op_t* selective_sync(blockstore_op_t *op);
     int split_stab_op(blockstore_op_t *op, std::function<int(obj_ver_id v)> decider);

src/blockstore_rollback.cpp

@@ -162,6 +162,7 @@ void blockstore_impl_t::mark_rolled_back(const obj_ver_id & ov)
                     unstable_writes.erase(unstab_it);
                 else
                     unstab_it->second = max_unstable;
+                unstable_count_changed = true;
             }
         }
     }

src/blockstore_stable.cpp

@@ -412,11 +412,40 @@ resume_4:
     return 2;
 }
 
-void blockstore_impl_t::mark_stable(const obj_ver_id & v, bool forget_dirty)
+void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
 {
     auto dirty_it = dirty_db.find(v);
     if (dirty_it != dirty_db.end())
     {
+        if (IS_INSTANT(dirty_it->second.state))
+        {
+            // 'Instant' (non-EC) operations may complete and try to become stable out of order. Prevent it.
+            auto back_it = dirty_it;
+            while (back_it != dirty_db.begin())
+            {
+                back_it--;
+                if (back_it->first.oid != v.oid)
+                {
+                    break;
+                }
+                if (!IS_STABLE(back_it->second.state))
+                {
+                    // There are preceding unstable versions, can't flush <v>
+                    return;
+                }
+            }
+            while (true)
+            {
+                dirty_it++;
+                if (dirty_it == dirty_db.end() || dirty_it->first.oid != v.oid ||
+                    !IS_SYNCED(dirty_it->second.state))
+                {
+                    dirty_it--;
+                    break;
+                }
+                v.version = dirty_it->first.version;
+            }
+        }
         while (1)
         {
             bool was_stable = IS_STABLE(dirty_it->second.state);
@@ -508,5 +537,6 @@ void blockstore_impl_t::mark_stable(const obj_ver_id & v, bool forget_dirty)
         unstab_it->second <= v.version)
     {
         unstable_writes.erase(unstab_it);
+        unstable_count_changed = true;
     }
 }

src/blockstore_sync.cpp

@@ -85,8 +85,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
                 left--;
                 auto & dirty_entry = dirty_db.at(sbw);
                 uint64_t dyn_size = dsk.dirty_dyn_size(dirty_entry.offset, dirty_entry.len);
-                if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
-                    (unstable_writes.size()+unstable_unsynced)*journal.block_size))
+                if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size, 0))
                 {
                     return 0;
                 }

src/blockstore_write.cpp

@@ -593,7 +593,7 @@ resume_4:
 #endif
         bool is_big = (dirty_it->second.state & BS_ST_TYPE_MASK) == BS_ST_BIG_WRITE;
         bool imm = is_big ? (immediate_commit == IMMEDIATE_ALL) : (immediate_commit != IMMEDIATE_NONE);
-        bool is_instant = ((dirty_it->second.state & BS_ST_TYPE_MASK) == BS_ST_DELETE || (dirty_it->second.state & BS_ST_INSTANT));
+        bool is_instant = IS_INSTANT(dirty_it->second.state);
         if (imm)
         {
             auto & unstab = unstable_writes[op->oid];

src/cluster_client.cpp

@@ -6,7 +6,7 @@
 #include "cluster_client_impl.h"
 #include "http_client.h" // json_is_true
 
-cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config)
+cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config)
 {
     wb = new writeback_cache_t();
 
@@ -352,13 +352,13 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
     // up_wait_retry_interval
     up_wait_retry_interval = config["up_wait_retry_interval"].uint64_value();
     if (!up_wait_retry_interval)
-    {
-        up_wait_retry_interval = 500;
-    }
-    else if (up_wait_retry_interval < 50)
     {
         up_wait_retry_interval = 50;
     }
+    else if (up_wait_retry_interval < 10)
+    {
+        up_wait_retry_interval = 10;
+    }
     // log_level
     log_level = config["log_level"].uint64_value();
     msgr.parse_config(config);
@@ -534,7 +534,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) &&
-        !op->version /* FIXME no CAS writeback */)
+        !op->version /* no CAS writeback */)
     {
         if (wb->writebacks_active >= client_max_writeback_iodepth)
         {
@@ -555,7 +555,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     }
     if (op->opcode == OSD_OP_WRITE && !(op->flags & OP_IMMEDIATE_COMMIT))
     {
-        if (!(op->flags & OP_FLUSH_BUFFER))
+        if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */)
         {
             wb->copy_write(op, CACHE_WRITTEN);
         }
@@ -1161,7 +1161,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
                 );
             }
         }
-        else
+        else if (log_level > 0)
         {
             fprintf(
                 stderr, "%s operation failed on OSD %lu: retval=%ld (expected %d)\n",

src/cluster_client.h

@@ -121,7 +121,7 @@ public:
     json11::Json::object cli_config, file_config, etcd_global_config;
     json11::Json::object config;
 
-    cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json & config);
+    cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config);
     ~cluster_client_t();
     void execute(cluster_op_t *op);
     void execute_raw(osd_num_t osd_num, osd_op_t *op);

src/kv_cli.cpp

@@ -0,0 +1,401 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+//
+// Vitastor shared key/value database test CLI
+
+#define _XOPEN_SOURCE
+#include <limits.h>
+
+#include <netinet/tcp.h>
+#include <sys/epoll.h>
+#include <unistd.h>
+#include <fcntl.h>
+//#include <signal.h>
+
+#include "epoll_manager.h"
+#include "str_util.h"
+#include "kv_db.h"
+
+const char *exe_name = NULL;
+
+class kv_cli_t
+{
+public:
+    kv_dbw_t *db = NULL;
+    ring_loop_t *ringloop = NULL;
+    epoll_manager_t *epmgr = NULL;
+    cluster_client_t *cli = NULL;
+    bool interactive = false;
+    int in_progress = 0;
+    char *cur_cmd = NULL;
+    int cur_cmd_size = 0, cur_cmd_alloc = 0;
+    bool finished = false, eof = false;
+    json11::Json::object cfg;
+
+    ~kv_cli_t();
+
+    static json11::Json::object parse_args(int narg, const char *args[]);
+    void run(const json11::Json::object & cfg);
+    void read_cmd();
+    void next_cmd();
+    void handle_cmd(const std::string & cmd, std::function<void()> cb);
+};
+
+kv_cli_t::~kv_cli_t()
+{
+    if (cur_cmd)
+    {
+        free(cur_cmd);
+        cur_cmd = NULL;
+    }
+    cur_cmd_alloc = 0;
+    if (db)
+        delete db;
+    if (cli)
+    {
+        cli->flush();
+        delete cli;
+    }
+    if (epmgr)
+        delete epmgr;
+    if (ringloop)
+        delete ringloop;
+}
+
+json11::Json::object kv_cli_t::parse_args(int narg, const char *args[])
+{
+    json11::Json::object cfg;
+    for (int i = 1; i < narg; i++)
+    {
+        if (!strcmp(args[i], "-h") || !strcmp(args[i], "--help"))
+        {
+            printf(
+                "Vitastor Key/Value CLI\n"
+                "(c) Vitaliy Filippov, 2023+ (VNPL-1.1)\n"
+                "\n"
+                "USAGE: %s [--etcd_address ADDR] [OTHER OPTIONS]\n",
+                exe_name
+            );
+            exit(0);
+        }
+        else if (args[i][0] == '-' && args[i][1] == '-')
+        {
+            const char *opt = args[i]+2;
+            cfg[opt] = !strcmp(opt, "json") || i == narg-1 ? "1" : args[++i];
+        }
+    }
+    return cfg;
+}
+
+void kv_cli_t::run(const json11::Json::object & cfg)
+{
+    // Create client
+    ringloop = new ring_loop_t(512);
+    epmgr = new epoll_manager_t(ringloop);
+    cli = new cluster_client_t(ringloop, epmgr->tfd, cfg);
+    db = new kv_dbw_t(cli);
+    // Load image metadata
+    while (!cli->is_ready())
+    {
+        ringloop->loop();
+        if (cli->is_ready())
+            break;
+        ringloop->wait();
+    }
+    // Run
+    fcntl(0, F_SETFL, fcntl(0, F_GETFL, 0) | O_NONBLOCK);
+    try
+    {
+        epmgr->tfd->set_fd_handler(0, false, [this](int fd, int events)
+        {
+            if (events & EPOLLIN)
+            {
+                read_cmd();
+            }
+            if (events & EPOLLRDHUP)
+            {
+                epmgr->tfd->set_fd_handler(0, false, NULL);
+                finished = true;
+            }
+        });
+        interactive = true;
+        printf("> ");
+    }
+    catch (std::exception & e)
+    {
+        // Can't add to epoll, STDIN is probably a file
+        read_cmd();
+    }
+    while (!finished)
+    {
+        ringloop->loop();
+        if (!finished)
+            ringloop->wait();
+    }
+    // Destroy the client
+    delete db;
+    db = NULL;
+    cli->flush();
+    delete cli;
+    delete epmgr;
+    delete ringloop;
+    cli = NULL;
+    epmgr = NULL;
+    ringloop = NULL;
+}
+
+void kv_cli_t::read_cmd()
+{
+    if (!cur_cmd_alloc)
+    {
+        cur_cmd_alloc = 65536;
+        cur_cmd = (char*)malloc_or_die(cur_cmd_alloc);
+    }
+    while (cur_cmd_size < cur_cmd_alloc)
+    {
+        int r = read(0, cur_cmd+cur_cmd_size, cur_cmd_alloc-cur_cmd_size);
+        if (r < 0 && errno != EAGAIN)
+            fprintf(stderr, "Error reading from stdin: %s\n", strerror(errno));
+        if (r > 0)
+            cur_cmd_size += r;
+        if (r == 0)
+            eof = true;
+        if (r <= 0)
+            break;
+    }
+    next_cmd();
+}
+
+void kv_cli_t::next_cmd()
+{
+    if (in_progress > 0)
+    {
+        return;
+    }
+    int pos = 0;
+    for (; pos < cur_cmd_size; pos++)
+    {
+        if (cur_cmd[pos] == '\n' || cur_cmd[pos] == '\r')
+        {
+            auto cmd = trim(std::string(cur_cmd, pos));
+            pos++;
+            memmove(cur_cmd, cur_cmd+pos, cur_cmd_size-pos);
+            cur_cmd_size -= pos;
+            in_progress++;
+            handle_cmd(cmd, [this]()
+            {
+                in_progress--;
+                if (interactive)
+                    printf("> ");
+                next_cmd();
+                if (!in_progress)
+                    read_cmd();
+            });
+            break;
+        }
+    }
+    if (eof && !in_progress)
+    {
+        finished = true;
+    }
+}
+
+void kv_cli_t::handle_cmd(const std::string & cmd, std::function<void()> cb)
+{
+    if (cmd == "")
+    {
+        cb();
+        return;
+    }
+    auto pos = cmd.find_first_of(" \t");
+    if (pos != std::string::npos)
+    {
+        while (pos < cmd.size()-1 && (cmd[pos+1] == ' ' || cmd[pos+1] == '\t'))
+            pos++;
+    }
+    auto opname = strtolower(pos == std::string::npos ? cmd : cmd.substr(0, pos));
+    if (opname == "open")
+    {
+        uint64_t pool_id = 0;
+        inode_t inode_id = 0;
+        uint32_t kv_block_size = 0;
+        int scanned = sscanf(cmd.c_str() + pos+1, "%lu %lu %u", &pool_id, &inode_id, &kv_block_size);
+        if (scanned == 2)
+        {
+            kv_block_size = 4096;
+        }
+        if (scanned < 2 || !pool_id || !inode_id || !kv_block_size || (kv_block_size & (kv_block_size-1)) != 0)
+        {
+            fprintf(stderr, "Usage: open <pool_id> <inode_id> [block_size]. Block size must be a power of 2. Default is 4096.\n");
+            cb();
+            return;
+        }
+        cfg["kv_block_size"] = (uint64_t)kv_block_size;
+        db->open(INODE_WITH_POOL(pool_id, inode_id), cfg, [=](int res)
+        {
+            if (res < 0)
+                fprintf(stderr, "Error opening index: %s (code %d)\n", strerror(-res), res);
+            else
+                printf("Index opened. Current size: %lu bytes\n", db->get_size());
+            cb();
+        });
+    }
+    else if (opname == "config")
+    {
+        auto pos2 = cmd.find_first_of(" \t", pos+1);
+        if (pos2 == std::string::npos)
+        {
+            fprintf(stderr, "Usage: config <property> <value>\n");
+            cb();
+            return;
+        }
+        auto key = trim(cmd.substr(pos+1, pos2-pos-1));
+        auto value = parse_size(trim(cmd.substr(pos2+1)));
+        if (key != "kv_memory_limit" &&
+            key != "kv_allocate_blocks" &&
+            key != "kv_evict_max_misses" &&
+            key != "kv_evict_attempts_per_level" &&
+            key != "kv_evict_unused_age" &&
+            key != "kv_log_level")
+        {
+            fprintf(
+                stderr, "Allowed properties: kv_memory_limit, kv_allocate_blocks,"
+                " kv_evict_max_misses, kv_evict_attempts_per_level, kv_evict_unused_age, kv_log_level\n"
+            );
+        }
+        else
+        {
+            cfg[key] = value;
+            db->set_config(cfg);
+        }
+        cb();
+    }
+    else if (opname == "get" || opname == "set" || opname == "del")
+    {
+        if (opname == "get" || opname == "del")
+        {
+            if (pos == std::string::npos)
+            {
+                fprintf(stderr, "Usage: %s <key>\n", opname.c_str());
+                cb();
+                return;
+            }
+            auto key = trim(cmd.substr(pos+1));
+            if (opname == "get")
+            {
+                db->get(key, [this, cb](int res, const std::string & value)
+                {
+                    if (res < 0)
+                        fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+                    else
+                    {
+                        write(1, value.c_str(), value.size());
+                        write(1, "\n", 1);
+                    }
+                    cb();
+                });
+            }
+            else
+            {
+                db->del(key, [this, cb](int res)
+                {
+                    if (res < 0)
+                        fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+                    else
+                        printf("OK\n");
+                    cb();
+                });
+            }
+        }
+        else
+        {
+            auto pos2 = cmd.find_first_of(" \t", pos+1);
+            if (pos2 == std::string::npos)
+            {
+                fprintf(stderr, "Usage: set <key> <value>\n");
+                cb();
+                return;
+            }
+            auto key = trim(cmd.substr(pos+1, pos2-pos-1));
+            auto value = trim(cmd.substr(pos2+1));
+            db->set(key, value, [this, cb](int res)
+            {
+                if (res < 0)
+                    fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+                else
+                    printf("OK\n");
+                cb();
+            });
+        }
+    }
+    else if (opname == "list")
+    {
+        std::string start, end;
+        if (pos != std::string::npos)
+        {
+            auto pos2 = cmd.find_first_of(" \t", pos+1);
+            if (pos2 != std::string::npos)
+            {
+                start = trim(cmd.substr(pos+1, pos2-pos-1));
+                end = trim(cmd.substr(pos2+1));
+            }
+            else
+            {
+                start = trim(cmd.substr(pos+1));
+            }
+        }
+        void *handle = db->list_start(start);
+        db->list_next(handle, [=](int res, const std::string & key, const std::string & value)
+        {
+            if (res < 0)
+            {
+                if (res != -ENOENT)
+                {
+                    fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+                }
+                db->list_close(handle);
+                cb();
+            }
+            else
+            {
+                printf("%s = %s\n", key.c_str(), value.c_str());
+                db->list_next(handle, NULL);
+            }
+        });
+    }
+    else if (opname == "close")
+    {
+        db->close([=]()
+        {
+            printf("Index closed\n");
+            cb();
+        });
+    }
+    else if (opname == "quit" || opname == "q")
+    {
+        ::close(0);
+        finished = true;
+    }
+    else
+    {
+        fprintf(
+            stderr, "Unknown operation: %s. Supported operations:\n"
+            "open <pool_id> <inode_id> [block_size]\n"
+            "config <property> <value>\n"
+            "get <key>\nset <key> <value>\ndel <key>\nlist [<start> [end]]\n"
+            "close\nquit\n", opname.c_str()
+        );
+        cb();
+    }
+}
+
+int main(int narg, const char *args[])
+{
+    setvbuf(stdout, NULL, _IONBF, 0);
+    setvbuf(stderr, NULL, _IONBF, 0);
+    exe_name = args[0];
+    kv_cli_t *p = new kv_cli_t();
+    p->run(kv_cli_t::parse_args(narg, args));
+    delete p;
+    return 0;
+}

src/kv_db.h

@@ -0,0 +1,36 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+//
+// Vitastor shared key/value database
+// Parallel optimistic B-Tree O:-)
+
+#pragma once
+
+#include "cluster_client.h"
+
+struct kv_db_t;
+
+struct kv_dbw_t
+{
+    kv_dbw_t(cluster_client_t *cli);
+    ~kv_dbw_t();
+
+    void open(inode_t inode_id, json11::Json cfg, std::function<void(int)> cb);
+    void set_config(json11::Json cfg);
+    void close(std::function<void()> cb);
+
+    uint64_t get_size();
+
+    void get(const std::string & key, std::function<void(int res, const std::string & value)> cb,
+        bool allow_old_cached = false);
+    void set(const std::string & key, const std::string & value, std::function<void(int res)> cb,
+        std::function<bool(int res, const std::string & value)> cas_compare = NULL);
+    void del(const std::string & key, std::function<void(int res)> cb,
+        std::function<bool(int res, const std::string & value)> cas_compare = NULL);
+
+    void* list_start(const std::string & start);
+    void list_next(void *handle, std::function<void(int res, const std::string & key, const std::string & value)> cb);
+    void list_close(void *handle);
+
+    kv_db_t *db;
+};

src/kv_stress.cpp

@@ -0,0 +1,697 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+//
+// Vitastor shared key/value database stress tester / benchmark
+
+#define _XOPEN_SOURCE
+#include <limits.h>
+
+#include <netinet/tcp.h>
+#include <sys/epoll.h>
+#include <unistd.h>
+#include <fcntl.h>
+//#include <signal.h>
+
+#include "epoll_manager.h"
+#include "str_util.h"
+#include "kv_db.h"
+
+const char *exe_name = NULL;
+
+struct kv_test_listing_t
+{
+    uint64_t count = 0, done = 0;
+    void *handle = NULL;
+    std::string next_after;
+    std::set<std::string> inflights;
+    timespec tv_begin;
+    bool error = false;
+};
+
+struct kv_test_lat_t
+{
+    const char *name = NULL;
+    uint64_t usec = 0, count = 0;
+};
+
+struct kv_test_stat_t
+{
+    kv_test_lat_t get, add, update, del, list;
+    uint64_t list_keys = 0;
+};
+
+class kv_test_t
+{
+public:
+    // Config
+    json11::Json::object kv_cfg;
+    std::string key_prefix, key_suffix;
+    uint64_t inode_id = 0;
+    uint64_t op_count = 1000000;
+    uint64_t runtime_sec = 0;
+    uint64_t parallelism = 4;
+    uint64_t reopen_prob = 1;
+    uint64_t get_prob = 30000;
+    uint64_t add_prob = 20000;
+    uint64_t update_prob = 20000;
+    uint64_t del_prob = 5000;
+    uint64_t list_prob = 300;
+    uint64_t min_key_len = 10;
+    uint64_t max_key_len = 70;
+    uint64_t min_value_len = 50;
+    uint64_t max_value_len = 300;
+    uint64_t min_list_count = 10;
+    uint64_t max_list_count = 1000;
+    uint64_t print_stats_interval = 1;
+    bool json_output = false;
+    uint64_t log_level = 1;
+    bool trace = false;
+    bool stop_on_error = false;
+    // FIXME: Multiple clients
+    kv_test_stat_t stat, prev_stat;
+    timespec prev_stat_time, start_stat_time;
+
+    // State
+    kv_dbw_t *db = NULL;
+    ring_loop_t *ringloop = NULL;
+    epoll_manager_t *epmgr = NULL;
+    cluster_client_t *cli = NULL;
+    ring_consumer_t consumer;
+    bool finished = false;
+    uint64_t total_prob = 0;
+    uint64_t ops_sent = 0, ops_done = 0;
+    int stat_timer_id = -1;
+    int in_progress = 0;
+    bool reopening = false;
+    std::set<kv_test_listing_t*> listings;
+    std::set<std::string> changing_keys;
+    std::map<std::string, std::string> values;
+
+    ~kv_test_t();
+
+    static json11::Json::object parse_args(int narg, const char *args[]);
+    void parse_config(json11::Json cfg);
+    void run(json11::Json cfg);
+    void loop();
+    void print_stats(kv_test_stat_t & prev_stat, timespec & prev_stat_time);
+    void print_total_stats();
+    void start_change(const std::string & key);
+    void stop_change(const std::string & key);
+    void add_stat(kv_test_lat_t & stat, timespec tv_begin);
+};
+
+kv_test_t::~kv_test_t()
+{
+    if (db)
+        delete db;
+    if (cli)
+    {
+        cli->flush();
+        delete cli;
+    }
+    if (epmgr)
+        delete epmgr;
+    if (ringloop)
+        delete ringloop;
+}
+
+json11::Json::object kv_test_t::parse_args(int narg, const char *args[])
+{
+    json11::Json::object cfg;
+    for (int i = 1; i < narg; i++)
+    {
+        if (!strcmp(args[i], "-h") || !strcmp(args[i], "--help"))
+        {
+            printf(
+                "Vitastor Key/Value DB stress tester / benchmark\n"
+                "(c) Vitaliy Filippov, 2023+ (VNPL-1.1)\n"
+                "\n"
+                "USAGE: %s --pool_id POOL_ID --inode_id INODE_ID [OPTIONS]\n"
+                "  --op_count 1000000\n"
+                "    Total operations to run during test. 0 means unlimited\n"
+                "  --key_prefix \"\"\n"
+                "    Prefix for all keys read or written (to avoid collisions)\n"
+                "  --key_suffix \"\"\n"
+                "    Suffix for all keys read or written (to avoid collisions, but scan all DB)\n"
+                "  --runtime 0\n"
+                "    Run for this number of seconds. 0 means unlimited\n"
+                "  --parallelism 4\n"
+                "    Run this number of operations in parallel\n"
+                "  --get_prob 30000\n"
+                "    Fraction of key retrieve operations\n"
+                "  --add_prob 20000\n"
+                "    Fraction of key addition operations\n"
+                "  --update_prob 20000\n"
+                "    Fraction of key update operations\n"
+                "  --del_prob 30000\n"
+                "    Fraction of key delete operations\n"
+                "  --list_prob 300\n"
+                "    Fraction of listing operations\n"
+                "  --min_key_len 10\n"
+                "    Minimum key size in bytes\n"
+                "  --max_key_len 70\n"
+                "    Maximum key size in bytes\n"
+                "  --min_value_len 50\n"
+                "    Minimum value size in bytes\n"
+                "  --max_value_len 300\n"
+                "    Maximum value size in bytes\n"
+                "  --min_list_count 10\n"
+                "    Minimum number of keys read in listing (0 = all keys)\n"
+                "  --max_list_count 1000\n"
+                "    Maximum number of keys read in listing\n"
+                "  --print_stats 1\n"
+                "    Print operation statistics every this number of seconds\n"
+                "  --json\n"
+                "    JSON output\n"
+                "  --stop_on_error 0\n"
+                "    Stop on first execution error, mismatch, lost key or extra key during listing\n"
+                "  --kv_memory_limit 128M\n"
+                "    Maximum memory to use for vitastor-kv index cache\n"
+                "  --kv_allocate_blocks 4\n"
+                "    Number of PG blocks used for new tree block allocation in parallel\n"
+                "  --kv_evict_max_misses 10\n"
+                "    Eviction algorithm parameter: retry eviction from another random spot\n"
+                "    if this number of keys is used currently or was used recently\n"
+                "  --kv_evict_attempts_per_level 3\n"
+                "    Retry eviction at most this number of times per tree level, starting\n"
+                "    with bottom-most levels\n"
+                "  --kv_evict_unused_age 1000\n"
+                "    Evict only keys unused during this number of last operations\n"
+                "  --kv_log_level 1\n"
+                "    Log level. 0 = errors, 1 = warnings, 10 = trace operations\n",
+                exe_name
+            );
+            exit(0);
+        }
+        else if (args[i][0] == '-' && args[i][1] == '-')
+        {
+            const char *opt = args[i]+2;
+            cfg[opt] = !strcmp(opt, "json") || i == narg-1 ? "1" : args[++i];
+        }
+    }
+    return cfg;
+}
+
+void kv_test_t::parse_config(json11::Json cfg)
+{
+    inode_id = INODE_WITH_POOL(cfg["pool_id"].uint64_value(), cfg["inode_id"].uint64_value());
+    if (cfg["op_count"].uint64_value() > 0)
+        op_count = cfg["op_count"].uint64_value();
+    key_prefix = cfg["key_prefix"].string_value();
+    key_suffix = cfg["key_suffix"].string_value();
+    if (cfg["runtime"].uint64_value() > 0)
+        runtime_sec = cfg["runtime"].uint64_value();
+    if (cfg["parallelism"].uint64_value() > 0)
+        parallelism = cfg["parallelism"].uint64_value();
+    if (!cfg["reopen_prob"].is_null())
+        reopen_prob = cfg["reopen_prob"].uint64_value();
+    if (!cfg["get_prob"].is_null())
+        get_prob = cfg["get_prob"].uint64_value();
+    if (!cfg["add_prob"].is_null())
+        add_prob = cfg["add_prob"].uint64_value();
+    if (!cfg["update_prob"].is_null())
+        update_prob = cfg["update_prob"].uint64_value();
+    if (!cfg["del_prob"].is_null())
+        del_prob = cfg["del_prob"].uint64_value();
+    if (!cfg["list_prob"].is_null())
+        list_prob = cfg["list_prob"].uint64_value();
+    if (!cfg["min_key_len"].is_null())
+        min_key_len = cfg["min_key_len"].uint64_value();
+    if (cfg["max_key_len"].uint64_value() > 0)
+        max_key_len = cfg["max_key_len"].uint64_value();
+    if (!cfg["min_value_len"].is_null())
+        min_value_len = cfg["min_value_len"].uint64_value();
+    if (cfg["max_value_len"].uint64_value() > 0)
+        max_value_len = cfg["max_value_len"].uint64_value();
+    if (!cfg["min_list_count"].is_null())
+        min_list_count = cfg["min_list_count"].uint64_value();
+    if (!cfg["max_list_count"].is_null())
+        max_list_count = cfg["max_list_count"].uint64_value();
+    if (!cfg["print_stats"].is_null())
+        print_stats_interval = cfg["print_stats"].uint64_value();
+    if (!cfg["json"].is_null())
+        json_output = true;
+    if (!cfg["stop_on_error"].is_null())
+        stop_on_error = cfg["stop_on_error"].bool_value();
+    if (!cfg["kv_memory_limit"].is_null())
+        kv_cfg["kv_memory_limit"] = cfg["kv_memory_limit"];
+    if (!cfg["kv_allocate_blocks"].is_null())
+        kv_cfg["kv_allocate_blocks"] = cfg["kv_allocate_blocks"];
+    if (!cfg["kv_evict_max_misses"].is_null())
+        kv_cfg["kv_evict_max_misses"] = cfg["kv_evict_max_misses"];
+    if (!cfg["kv_evict_attempts_per_level"].is_null())
+        kv_cfg["kv_evict_attempts_per_level"] = cfg["kv_evict_attempts_per_level"];
+    if (!cfg["kv_evict_unused_age"].is_null())
+        kv_cfg["kv_evict_unused_age"] = cfg["kv_evict_unused_age"];
+    if (!cfg["kv_log_level"].is_null())
+    {
+        log_level = cfg["kv_log_level"].uint64_value();
+        trace = log_level >= 10;
+        kv_cfg["kv_log_level"] = cfg["kv_log_level"];
+    }
+    total_prob = reopen_prob+get_prob+add_prob+update_prob+del_prob+list_prob;
+    stat.get.name = "get";
+    stat.add.name = "add";
+    stat.update.name = "update";
+    stat.del.name = "del";
+    stat.list.name = "list";
+}
+
+void kv_test_t::run(json11::Json cfg)
+{
+    srand48(time(NULL));
+    parse_config(cfg);
+    // Create client
+    ringloop = new ring_loop_t(512);
+    epmgr = new epoll_manager_t(ringloop);
+    cli = new cluster_client_t(ringloop, epmgr->tfd, cfg);
+    db = new kv_dbw_t(cli);
+    // Load image metadata
+    while (!cli->is_ready())
+    {
+        ringloop->loop();
+        if (cli->is_ready())
+            break;
+        ringloop->wait();
+    }
+    // Run
+    reopening = true;
+    db->open(inode_id, kv_cfg, [this](int res)
+    {
+        reopening = false;
+        if (res < 0)
+        {
+            fprintf(stderr, "ERROR: Open index: %d (%s)\n", res, strerror(-res));
+            exit(1);
+        }
+        if (trace)
+            printf("Index opened\n");
+        ringloop->wakeup();
+    });
+    consumer.loop = [this]() { loop(); };
+    ringloop->register_consumer(&consumer);
+    if (print_stats_interval)
+        stat_timer_id = epmgr->tfd->set_timer(print_stats_interval*1000, true, [this](int) { print_stats(prev_stat, prev_stat_time); });
+    clock_gettime(CLOCK_REALTIME, &start_stat_time);
+    prev_stat_time = start_stat_time;
+    while (!finished)
+    {
+        ringloop->loop();
+        if (!finished)
+            ringloop->wait();
+    }
+    if (stat_timer_id >= 0)
+        epmgr->tfd->clear_timer(stat_timer_id);
+    ringloop->unregister_consumer(&consumer);
+    // Print total stats
+    print_total_stats();
+    // Destroy the client
+    delete db;
+    db = NULL;
+    cli->flush();
+    delete cli;
+    delete epmgr;
+    delete ringloop;
+    cli = NULL;
+    epmgr = NULL;
+    ringloop = NULL;
+}
+
+static const char *base64_chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789@+/";
+
+std::string random_str(int len)
+{
+    std::string str;
+    str.resize(len);
+    for (int i = 0; i < len; i++)
+    {
+        str[i] = base64_chars[lrand48() % 64];
+    }
+    return str;
+}
+
+void kv_test_t::loop()
+{
+    if (reopening)
+    {
+        return;
+    }
+    if (ops_done >= op_count)
+    {
+        finished = true;
+    }
+    while (!finished && ops_sent < op_count && in_progress < parallelism)
+    {
+        uint64_t dice = (lrand48() % total_prob);
+        if (dice < reopen_prob)
+        {
+            reopening = true;
+            db->close([this]()
+            {
+                if (trace)
+                    printf("Index closed\n");
+                db->open(inode_id, kv_cfg, [this](int res)
+                {
+                    reopening = false;
+                    if (res < 0)
+                    {
+                        fprintf(stderr, "ERROR: Reopen index: %d (%s)\n", res, strerror(-res));
+                        finished = true;
+                        return;
+                    }
+                    if (trace)
+                        printf("Index reopened\n");
+                    ringloop->wakeup();
+                });
+            });
+            return;
+        }
+        else if (dice < reopen_prob+get_prob)
+        {
+            // get existing
+            auto key = random_str(max_key_len);
+            auto k_it = values.lower_bound(key);
+            if (k_it == values.end())
+                continue;
+            key = k_it->first;
+            if (changing_keys.find(key) != changing_keys.end())
+                continue;
+            in_progress++;
+            ops_sent++;
+            if (trace)
+                printf("get %s\n", key.c_str());
+            timespec tv_begin;
+            clock_gettime(CLOCK_REALTIME, &tv_begin);
+            db->get(key, [this, key, tv_begin](int res, const std::string & value)
+            {
+                add_stat(stat.get, tv_begin);
+                ops_done++;
+                in_progress--;
+                auto it = values.find(key);
+                if (res != (it == values.end() ? -ENOENT : 0))
+                {
+                    fprintf(stderr, "ERROR: get %s: %d (%s)\n", key.c_str(), res, strerror(-res));
+                    if (stop_on_error)
+                        exit(1);
+                }
+                else if (it != values.end() && value != it->second)
+                {
+                    fprintf(stderr, "ERROR: get %s: mismatch: %s vs %s\n", key.c_str(), value.c_str(), it->second.c_str());
+                    if (stop_on_error)
+                        exit(1);
+                }
+                ringloop->wakeup();
+            });
+        }
+        else if (dice < reopen_prob+get_prob+add_prob+update_prob)
+        {
+            bool is_add = false;
+            std::string key;
+            if (dice < reopen_prob+get_prob+add_prob)
+            {
+                // add
+                is_add = true;
+                uint64_t key_len = min_key_len + (max_key_len > min_key_len ? lrand48() % (max_key_len-min_key_len) : 0);
+                key = key_prefix + random_str(key_len) + key_suffix;
+            }
+            else
+            {
+                // update
+                key = random_str(max_key_len);
+                auto k_it = values.lower_bound(key);
+                if (k_it == values.end())
+                    continue;
+                key = k_it->first;
+            }
+            if (changing_keys.find(key) != changing_keys.end())
+                continue;
+            uint64_t value_len = min_value_len + (max_value_len > min_value_len ? lrand48() % (max_value_len-min_value_len) : 0);
+            auto value = random_str(value_len);
+            start_change(key);
+            ops_sent++;
+            in_progress++;
+            if (trace)
+                printf("set %s = %s\n", key.c_str(), value.c_str());
+            timespec tv_begin;
+            clock_gettime(CLOCK_REALTIME, &tv_begin);
+            db->set(key, value, [this, key, value, tv_begin, is_add](int res)
+            {
+                add_stat(is_add ? stat.add : stat.update, tv_begin);
+                stop_change(key);
+                ops_done++;
+                in_progress--;
+                if (res != 0)
+                {
+                    fprintf(stderr, "ERROR: set %s = %s: %d (%s)\n", key.c_str(), value.c_str(), res, strerror(-res));
+                    if (stop_on_error)
+                        exit(1);
+                }
+                else
+                {
+                    values[key] = value;
+                }
+                ringloop->wakeup();
+            }, NULL);
+        }
+        else if (dice < reopen_prob+get_prob+add_prob+update_prob+del_prob)
+        {
+            // delete
+            auto key = random_str(max_key_len);
+            auto k_it = values.lower_bound(key);
+            if (k_it == values.end())
+                continue;
+            key = k_it->first;
+            if (changing_keys.find(key) != changing_keys.end())
+                continue;
+            start_change(key);
+            ops_sent++;
+            in_progress++;
+            if (trace)
+                printf("del %s\n", key.c_str());
+            timespec tv_begin;
+            clock_gettime(CLOCK_REALTIME, &tv_begin);
+            db->del(key, [this, key, tv_begin](int res)
+            {
+                add_stat(stat.del, tv_begin);
+                stop_change(key);
+                ops_done++;
+                in_progress--;
+                if (res != 0)
+                {
+                    fprintf(stderr, "ERROR: del %s: %d (%s)\n", key.c_str(), res, strerror(-res));
+                    if (stop_on_error)
+                        exit(1);
+                }
+                else
+                {
+                    values.erase(key);
+                }
+                ringloop->wakeup();
+            }, NULL);
+        }
+        else if (dice < reopen_prob+get_prob+add_prob+update_prob+del_prob+list_prob)
+        {
+            // list
+            ops_sent++;
+            in_progress++;
+            auto key = random_str(max_key_len);
+            auto lst = new kv_test_listing_t;
+            auto k_it = values.lower_bound(key);
+            lst->count = min_list_count + (max_list_count > min_list_count ? lrand48() % (max_list_count-min_list_count) : 0);
+            lst->handle = db->list_start(k_it == values.begin() ? key_prefix : key);
+            lst->next_after = k_it == values.begin() ? key_prefix : key;
+            lst->inflights = changing_keys;
+            listings.insert(lst);
+            if (trace)
+                printf("list from %s\n", key.c_str());
+            clock_gettime(CLOCK_REALTIME, &lst->tv_begin);
+            db->list_next(lst->handle, [this, lst](int res, const std::string & key, const std::string & value)
+            {
+                if (log_level >= 11)
+                    printf("list: %s = %s\n", key.c_str(), value.c_str());
+                if (res >= 0 && key_prefix.size() && (key.size() < key_prefix.size() ||
+                    key.substr(0, key_prefix.size()) != key_prefix))
+                {
+                    // stop at this key
+                    res = -ENOENT;
+                }
+                if (res < 0 || (lst->count > 0 && lst->done >= lst->count))
+                {
+                    add_stat(stat.list, lst->tv_begin);
+                    if (res == 0)
+                    {
+                        // ok (done >= count)
+                    }
+                    else if (res != -ENOENT)
+                    {
+                        fprintf(stderr, "ERROR: list: %d (%s)\n", res, strerror(-res));
+                        lst->error = true;
+                    }
+                    else
+                    {
+                        auto k_it = lst->next_after == "" ? values.begin() : values.upper_bound(lst->next_after);
+                        while (k_it != values.end())
+                        {
+                            while (k_it != values.end() && lst->inflights.find(k_it->first) != lst->inflights.end())
+                                k_it++;
+                            if (k_it != values.end())
+                            {
+                                fprintf(stderr, "ERROR: list: missing key %s\n", (k_it++)->first.c_str());
+                                lst->error = true;
+                            }
+                        }
+                    }
+                    if (lst->error && stop_on_error)
+                        exit(1);
+                    ops_done++;
+                    in_progress--;
+                    db->list_close(lst->handle);
+                    delete lst;
+                    listings.erase(lst);
+                    ringloop->wakeup();
+                }
+                else
+                {
+                    stat.list_keys++;
+                    // Do not check modified keys in listing
+                    // Listing may return their old or new state
+                    if ((!key_suffix.size() || key.size() >= key_suffix.size() &&
+                        key.substr(key.size()-key_suffix.size()) == key_suffix) &&
+                        lst->inflights.find(key) == lst->inflights.end())
+                    {
+                        lst->done++;
+                        auto k_it = lst->next_after == "" ? values.begin() : values.upper_bound(lst->next_after);
+                        while (true)
+                        {
+                            while (k_it != values.end() && lst->inflights.find(k_it->first) != lst->inflights.end())
+                            {
+                                k_it++;
+                            }
+                            if (k_it == values.end() || k_it->first > key)
+                            {
+                                fprintf(stderr, "ERROR: list: extra key %s\n", key.c_str());
+                                lst->error = true;
+                                break;
+                            }
+                            else if (k_it->first < key)
+                            {
+                                fprintf(stderr, "ERROR: list: missing key %s\n", k_it->first.c_str());
+                                lst->error = true;
+                                lst->next_after = k_it->first;
+                                k_it++;
+                            }
+                            else
+                            {
+                                if (k_it->second != value)
+                                {
+                                    fprintf(stderr, "ERROR: list: mismatch: %s = %s but should be %s\n",
+                                        key.c_str(), value.c_str(), k_it->second.c_str());
+                                    lst->error = true;
+                                }
+                                lst->next_after = k_it->first;
+                                break;
+                            }
+                        }
+                    }
+                    db->list_next(lst->handle, NULL);
+                }
+            });
+        }
+    }
+}
+
+void kv_test_t::add_stat(kv_test_lat_t & stat, timespec tv_begin)
+{
+    timespec tv_end;
+    clock_gettime(CLOCK_REALTIME, &tv_end);
+    int64_t usec = (tv_end.tv_sec - tv_begin.tv_sec)*1000000 +
+        (tv_end.tv_nsec - tv_begin.tv_nsec)/1000;
+    if (usec > 0)
+    {
+        stat.usec += usec;
+        stat.count++;
+    }
+}
+
+void kv_test_t::print_stats(kv_test_stat_t & prev_stat, timespec & prev_stat_time)
+{
+    timespec cur_stat_time;
+    clock_gettime(CLOCK_REALTIME, &cur_stat_time);
+    int64_t usec = (cur_stat_time.tv_sec - prev_stat_time.tv_sec)*1000000 +
+        (cur_stat_time.tv_nsec - prev_stat_time.tv_nsec)/1000;
+    if (usec > 0)
+    {
+        kv_test_lat_t *lats[] = { &stat.get, &stat.add, &stat.update, &stat.del, &stat.list };
+        kv_test_lat_t *prev[] = { &prev_stat.get, &prev_stat.add, &prev_stat.update, &prev_stat.del, &prev_stat.list };
+        if (!json_output)
+        {
+            char buf[128] = { 0 };
+            for (int i = 0; i < sizeof(lats)/sizeof(lats[0]); i++)
+            {
+                snprintf(buf, sizeof(buf)-1, "%.1f %s/s (%lu us)", (lats[i]->count-prev[i]->count)*1000000.0/usec,
+                    lats[i]->name, (lats[i]->usec-prev[i]->usec)/(lats[i]->count-prev[i]->count > 0 ? lats[i]->count-prev[i]->count : 1));
+                int k;
+                for (k = strlen(buf); k < strlen(lats[i]->name)+21; k++)
+                    buf[k] = ' ';
+                buf[k] = 0;
+                printf("%s", buf);
+            }
+            printf("\n");
+        }
+        else
+        {
+            int64_t runtime = (cur_stat_time.tv_sec - start_stat_time.tv_sec)*1000000 +
+                (cur_stat_time.tv_nsec - start_stat_time.tv_nsec)/1000;
+            printf("{\"runtime\":%.1f", (double)runtime/1000000.0);
+            for (int i = 0; i < sizeof(lats)/sizeof(lats[0]); i++)
+            {
+                if (lats[i]->count > prev[i]->count)
+                {
+                    printf(
+                        ",\"%s\":{\"avg\":{\"iops\":%.1f,\"usec\":%lu},\"total\":{\"count\":%lu,\"usec\":%lu}}",
+                        lats[i]->name, (lats[i]->count-prev[i]->count)*1000000.0/usec,
+                        (lats[i]->usec-prev[i]->usec)/(lats[i]->count-prev[i]->count),
+                        lats[i]->count, lats[i]->usec
+                    );
+                }
+            }
+            printf("}\n");
+        }
+    }
+    prev_stat = stat;
+    prev_stat_time = cur_stat_time;
+}
+
+void kv_test_t::print_total_stats()
+{
+    if (!json_output)
+        printf("Total:\n");
+    kv_test_stat_t start_stats;
+    timespec start_stat_time = this->start_stat_time;
+    print_stats(start_stats, start_stat_time);
+}
+
+void kv_test_t::start_change(const std::string & key)
+{
+    changing_keys.insert(key);
+    for (auto lst: listings)
+    {
+        lst->inflights.insert(key);
+    }
+}
+
+void kv_test_t::stop_change(const std::string & key)
+{
+    changing_keys.erase(key);
+}
+
+int main(int narg, const char *args[])
+{
+    setvbuf(stdout, NULL, _IONBF, 0);
+    setvbuf(stderr, NULL, _IONBF, 0);
+    exe_name = args[0];
+    kv_test_t *p = new kv_test_t();
+    p->run(kv_test_t::parse_args(narg, args));
+    delete p;
+    return 0;
+}

tests/test_scrub.sh

@@ -20,6 +20,9 @@ LD_PRELOAD="build/src/libfio_vitastor.so" \
     fio -thread -name=test -ioengine=build/src/libfio_vitastor.so -bs=1M -direct=1 -iodepth=4 \
         -mirror_file=./testdata/mirror.bin -end_fsync=1 -rw=write -etcd=$ETCD_URL -image=testimg
 
+# Save PG primary
+primary=$($ETCDCTL get --print-value-only /vitastor/config/pgs | jq -r '.items["1"]["1"].primary')
+
 # Intentionally corrupt OSD data and restart it
 zero_osd_pid=OSD${ZERO_OSD}_PID
 kill ${!zero_osd_pid}
@@ -34,6 +37,9 @@ start_osd $ZERO_OSD
 # Wait until start
 wait_up 10
 
+# Wait until PG is back on the same primary
+wait_condition 10 "$ETCDCTL"$' get --print-value-only /vitastor/config/pgs | jq -s -e \'.[0].items["1"]["1"].primary == "'$primary'"'"'"
+
 # Trigger scrub
 $ETCDCTL put /vitastor/pg/history/1/1 `$ETCDCTL get --print-value-only /vitastor/pg/history/1/1 | jq -s -c '(.[0] // {}) + {"next_scrub":1}'`