Compare commits

...

29 Commits

Author SHA1 Message Date
06a5d24240 Release 3.2.1
- Blockstores
  - Add more asynchrony to the chaos test to uncover more issues...
  - ...and fix them (~6 in the old store, ~9 in the new store)
  - Fix size check for journal_offset when sharing meta device (thanks Changwei Ye)
- Network layer
  - Correctly handle out-of-order zero-copy notifications (rare but possible in theory)
  - Fix double checksumming of header in proto_checksums=full mode
  - Add rxbounce option for Windows guests with TCP and enabled checksums
  - Close stopped peer sockets early to prevent hung zerocopy notifications and slow failover (~15 min instead of ~10 sec)
- OSD
  - Do not die on local blockstore modification errors (irrelevant since added write error handling)
- CSI
  - Fix VitastorFS NFS daemon lifecycle (thanks Dmitry Naumov)
  - Use a newer qemu-storage-daemon in CSI
- vitastor-cli
  - Add `dd` opool option
  - Fix `rm` with "inverse" optimisation incorrectly removing layer data before rename
    possibly leading to incorrect reads from child images during removal of an older-than-latest snapshot
  - Fix a possible hang in `merge` and derived commands (`rm`, `flatten`)
  - Fix mistakes in the `serve` OpenAPI spec (thanks Stepan Rabotkin)
  - Fix possible segfaults in `serve` due to invalid request parsing (thanks Vincent Schweiger)
  - Fix `serve` no-HTTPS mode (thanks both :))
- Build
  - Fix non-x86 builds failing due to unsupported xxhash x86dispatch
  - Forcibly enable assertions in all builds to remove build profile dependency
  - Fix QEMU driver build with QEMU < 10.2
2026-09-13 11:11:12 +03:00
a6a99a5967 Close sockets to prevent hung zerocopy notifications - with surrogate client IDs it's safe now 2026-09-11 02:05:08 +03:00
eba8ca7192 Add rxbounce option for Windows guests 2026-09-11 02:04:56 +03:00
Dmitry Naumov
e1d9121870 Fix VitastorFS NFS daemon lifecycle in CSI 2026-09-11 00:46:57 +03:00
72620ebee5 Store v1: Make sync wait for deletions which erase unsynced writes
Found by test_blockstore_chaos --seed 1487 and 534428711
2026-09-09 10:07:04 +03:00
d0e3ee4c2a Store v2: Repeat recheck for entries behind a dropped BIG_INTENT
The following sequence is possible:
big_write - small_write (unfinished) - big_intent (unfinished)

In this case we should recheck both overwrites. But we can't just always recheck
small_writes before big_intents because they're garbage if the big_intent is finished.
So recheck them only if the big_intent itself is marked as unfinished and dropped.

Found by test_blockstore_chaos --seed 165428650 and 214734465
2026-09-09 10:02:46 +03:00
328bf0cba1 Store v2: Don't skip checksums for unfinished overwrites during recheck
Found by test_blockstore_chaos --seed 370644250
2026-09-09 10:02:17 +03:00
16ce85d685 Store v2: Do not allow intent writes over unfinished compaction big_write
Found by test_blockstore_chaos --seed 1034655488
2026-09-09 10:00:45 +03:00
606dd820cd Store v2: Fix compaction versions on repeated commits with unstable entries between them
big v1 -> small v2 -> commit v1 -> small v3 -> commit v2 was writing a compaction entry
with version=1 but with v2 actually compacted in it.

Found by test_blockstore_chaos --seed 255885411
2026-09-09 10:00:13 +03:00
f9f0b35311 Add vitastor-cli dd opool option 2026-09-08 00:56:21 +03:00
778a3469b4 Fix double checksumming of header in proto_checksums=full mode 2026-09-08 00:25:12 +03:00
a8ed25ddf5 Store v1: free dyn_data when WRITE_STABLE over WRITE is refused
Signed-off-by: Yury Luneff <koolkhel@mail.ru>
2026-09-07 00:39:14 +03:00
49877a0fa9 Fix too long osd_peering_pg test output 2026-09-07 00:37:03 +03:00
4e11fd256e Fix printing make_test errors in CI on failure 2026-09-06 21:04:03 +03:00
1a1c338cc9 Clear write_op/write_ops in client destructor 2026-09-06 18:29:10 +03:00
7f88846a5d Replace send_free_ops + zc_free_list with a single send_free queue, handle out-of-order zerocopy completions correcty 2026-09-06 18:29:10 +03:00
d5f1f6967a Fix uninitialized op_enc 2026-09-06 18:29:04 +03:00
c486a14f6a Use vector ring_datas and free_ring_data in ringloop
(Also fixes cleanup of callbacks still queued on destruction)
2026-09-06 16:53:07 +03:00
df3c8cf72b Do not die on local blockstore modification errors 2026-09-06 16:53:07 +03:00
a8f5d7cafe Disable output buffering in test_cas 2026-09-06 14:50:25 +03:00
c5efed2061 Silence build warnings in std::regex 2026-09-06 14:09:04 +03:00
d5e56827c7 Print test_chaos output on failure 2026-09-06 11:24:03 +03:00
a788add518 Return EBUSY on ROLLBACK when already stable, like in the new store 2026-09-06 11:24:03 +03:00
d5ebc93e7d Add libvirt patch for 11.10 el10 2026-09-05 21:02:51 +03:00
9d4067bcf0 Add spec patch for qemu 10.1 el10 2026-09-05 20:37:19 +03:00
0be4cba1e6 Fix QEMU driver build with QEMU < 10.2 2026-09-05 20:06:52 +03:00
2179f6ab55 Use a newer QSD in CSI and fix CSI build 2026-09-05 17:46:35 +03:00
1dd4cbd7c3 Store v1: Fix deadlock when trying rollback a committed write (fuzzing) 2026-09-05 15:24:36 +03:00
174b36cf46 cli/rm: Move children of "inverse child" before removing its data 2026-09-05 14:32:29 +03:00
60 changed files with 1937 additions and 286 deletions

View File

@@ -62,8 +62,13 @@ jobs:
needs: build
container: ${{env.TEST_IMAGE}}:${{github.sha}}
steps:
# leak sanitizer sometimes crashes
- run: cd /root/vitastor/build && make -j16 build_tests test
- name: Run tests
id: test
run: cd /root/vitastor/build && make -j16 build_tests test
- name: Print logs
if: always() && steps.test.outcome == 'failure'
run: |
cat /root/vitastor/build/Testing/Temporary/LastTest.log
npm_lint:
runs-on: ubuntu-latest

View File

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 2.8...3.30)
project(vitastor)
set(VITASTOR_VERSION "3.2.0")
set(VITASTOR_VERSION "3.2.1")
include(CTest)

View File

@@ -37,8 +37,8 @@ RUN (echo deb http://vitastor.io/debian trixie main > /etc/apt/sources.list.d/vi
wget -q -O /etc/apt/trusted.gpg.d/vitastor.gpg https://vitastor.io/debian/pubkey.gpg && \
apt-get update && \
apt-get install -y vitastor-client ibverbs-providers && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.2%2Bds-2%2Bvitastor1/qemu-utils_10.0.2%2Bds-2%2Bvitastor1_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.2%2Bds-2%2Bvitastor1/qemu-block-extra_10.0.2%2Bds-2%2Bvitastor1_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2/qemu-utils_10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2/qemu-block-extra_10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2_amd64.deb && \
dpkg -x qemu-utils*.deb tmp1 && \
dpkg -x qemu-block-extra*.deb tmp1 && \
cp -a tmp1/usr/bin/qemu-storage-daemon /usr/bin/ && \

View File

@@ -25,8 +25,8 @@ RUN apt-get update && \
# NFS mount dependencies
nfs-common netbase \
# dependencies of qemu-storage-daemon
libnuma1 liburing2 libglib2.0-0 libfuse3-3 libaio1 libzstd1 libnettle8 \
libgmp10 libhogweed6 libp11-kit0 libidn2-0 libunistring2 libtasn1-6 libpcre2-8-0 libffi8 && \
libaio1t64 libc6 libfuse3-4 libglib2.0-0t64 libgmp10 libgnutls30t64 \
libhogweed6t64 libnettle8t64 libnuma1 libselinux1 liburing2 libzstd1 zlib1g && \
apt-get clean && \
(echo options nbd nbds_max=128 > /etc/modprobe.d/nbd.conf)
@@ -36,8 +36,8 @@ ADD deb /deb
RUN apt-get update && \
apt-get -y install /deb/vitastor-client_*.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-9.2.2%2Bds-1%2Bvitastor4/qemu-utils_9.2.2%2Bds-1%2Bvitastor4_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-9.2.2%2Bds-1%2Bvitastor4/qemu-block-extra_9.2.2%2Bds-1%2Bvitastor4_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2/qemu-utils_10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2_amd64.deb && \
wget https://vitastor.io/archive/qemu/qemu-trixie-10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2/qemu-block-extra_10.0.11%2Bds-0%2Bdeb13u1%2Bvitastor2_amd64.deb && \
dpkg -x qemu-utils*.deb tmp1 && \
dpkg -x qemu-block-extra*.deb tmp1 && \
cp -a tmp1/usr/bin/qemu-storage-daemon /usr/bin/ && \

View File

@@ -1,4 +1,4 @@
VITASTOR_VERSION ?= v3.2.0
VITASTOR_VERSION ?= v3.2.1
all: build push

View File

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

View File

@@ -87,7 +87,7 @@ spec:
- name: socket-dir
mountPath: /csi
- name: csi-attacher
image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.0
image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.1
args:
- "--v=5"
- "--csi-address=$(ADDRESS)"
@@ -121,7 +121,7 @@ spec:
privileged: true
capabilities:
add: ["SYS_ADMIN"]
image: vitalif/vitastor-csi:v3.2.0
image: vitalif/vitastor-csi:v3.2.1
args:
- "--node=$(NODE_ID)"
- "--endpoint=$(CSI_ENDPOINT)"

View File

@@ -7,6 +7,7 @@ require (
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
github.com/kubernetes-csi/csi-lib-utils v0.9.1
golang.org/x/net v0.7.0
golang.org/x/sys v0.5.0 // indirect
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
google.golang.org/grpc v1.33.1
google.golang.org/protobuf v1.24.0

View File

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

View File

@@ -8,7 +8,6 @@ import (
"encoding/json"
"fmt"
"strings"
"time"
"os"
"io/ioutil"

View File

@@ -96,18 +96,16 @@ func NewNodeServer(driver *Driver) *NodeServer
{
ns.restoreUblkDaemons()
}
if (ns.method == MOUNT_VDUSE || ns.method == MOUNT_UBLK)
ns.restoreNfsDaemons()
dur, err := time.ParseDuration(os.Getenv("RESTART_INTERVAL"))
if (err != nil)
{
dur, err := time.ParseDuration(os.Getenv("RESTART_INTERVAL"))
if (err != nil)
{
dur = 10 * time.Second
}
ns.restartInterval = dur
if (ns.restartInterval != time.Duration(0))
{
go ns.restarter()
}
dur = 10 * time.Second
}
ns.restartInterval = dur
if (ns.restartInterval != time.Duration(0))
{
go ns.restarter()
}
return ns
}
@@ -134,10 +132,8 @@ func (ns *NodeServer) unlockVolume(lockId string)
func (ns *NodeServer) restarter()
{
// Restart dead VDUSE daemons at regular intervals
// Otherwise volume I/O may hang in case of a qemu-storage-daemon crash
// Moreover, it may lead to a kernel panic of the kernel is configured to
// panic on hung tasks
// Restart dead storage daemons at regular intervals. Otherwise volume I/O
// may hang and may even lead to a kernel panic if panic_on_hung_task is set.
ticker := time.NewTicker(ns.restartInterval)
defer ticker.Stop()
for
@@ -151,6 +147,7 @@ func (ns *NodeServer) restarter()
{
ns.restoreUblkDaemons()
}
ns.restoreNfsDaemons()
}
}
@@ -329,15 +326,10 @@ func (ns *NodeServer) restoreNfsDaemons()
{
return
}
activeNFS, err := ns.listActiveNFS()
if (err != nil)
{
return
}
// Check all state files and try to restore active mounts
for _, stateFile := range stateFiles
{
ns.checkNfsState(stateFile, activeNFS)
ns.checkNfsState(stateFile)
}
}
@@ -363,24 +355,43 @@ func (ns *NodeServer) readNfsState(stateFile string, allowNotExists bool) (*NfsS
return &state, nil
}
func (ns *NodeServer) checkNfsState(stateFile string, activeNfs map[int][]string)
func (ns *NodeServer) checkNfsState(stateFile string)
{
// Read state file
// Read state file to determine the lock ID
state, err := ns.readNfsState(stateFile, false)
if (err != nil)
{
return
}
// Lock FS
ns.lockVolume(state.ConfigPath+":fs:"+state.FsName)
defer ns.unlockVolume(state.ConfigPath+":fs:"+state.FsName)
// Check if NFS at this port is still mounted
lockId := state.ConfigPath+":fs:"+state.FsName
ns.lockVolume(lockId)
defer ns.unlockVolume(lockId)
// Recheck state and mounts under the lock: publish/unpublish may have changed
// both while this recovery attempt was waiting for the lock.
state, err = ns.readNfsState(stateFile, true)
if (err != nil || state == nil)
{
return
}
activeNfs, err := ns.listActiveNFS()
if (err != nil)
{
return
}
pidFile := ns.stateDir + filepath.Base(stateFile)
pidFile = pidFile[0:len(pidFile)-5] + ".pid"
if (len(activeNfs[state.Port]) == 0)
{
// this is a stale state file, remove it
// This is a stale state file. Remove the root staging mount before
// stopping the daemon, otherwise a dead hidden NFS mount may remain.
klog.Warningf("state file %v contains stale mount at port %d, removing it", stateFile, state.Port)
err = mount.CleanupMountPoint(state.Path, ns.mounter, false)
if (err != nil)
{
klog.Errorf("failed to unmount stale NFS mount %v: %v", state.Path, err)
return
}
ns.stopNFS(stateFile, pidFile)
return
}
@@ -400,6 +411,7 @@ func (ns *NodeServer) checkNfsState(stateFile string, activeNfs map[int][]string
"--pidfile", pidFile,
"--bind", "127.0.0.1",
"--port", fmt.Sprintf("%d", state.Port),
"--config_path", state.ConfigPath,
"--fs", state.FsName,
"--pool", state.Pool,
"--portmap", "0",
@@ -725,6 +737,7 @@ func (ns *NodeServer) mountNFS(ctxVars map[string]string) (string, error)
"--pidfile", pidFile,
"--bind", "127.0.0.1",
"--port", "auto",
"--config_path", state.ConfigPath,
"--fs", state.FsName,
"--pool", state.Pool,
"--portmap", "0",
@@ -815,26 +828,38 @@ func (ns *NodeServer) listActiveNFS() (map[int][]string, error)
klog.Errorf("failed to list mounts: %v", err)
return nil, err
}
return collectActiveNFS(mounts), nil
}
func collectActiveNFS(mounts []mount.MountInfo) map[int][]string
{
activeNFS := make(map[int][]string)
for _, mount := range mounts
for _, mnt := range mounts
{
// Volume mounts always refer to subpaths
if (mount.FsType == "nfs" && mount.Root != "/")
// Only count PVC subdirectory mounts, not the root staging mount. Newer
// kernels expose the subdirectory in Source while older ones expose it
// in Root. The staging mount may also be visible through several bind
// aliases in the CSI container, so its mount point isn't a reliable
// discriminator.
if (mnt.FsType != "nfs" || !strings.HasPrefix(mnt.Source, "127.0.0.1:/") ||
(mnt.Source == "127.0.0.1:/" && mnt.Root == "/"))
{
for _, opt := range mount.MountOptions
continue
}
// NFS-specific options, including port, are per-superblock options.
for _, opt := range mnt.SuperOptions
{
if (strings.HasPrefix(opt, "port="))
{
if (strings.HasPrefix(opt, "port="))
port64, err := strconv.ParseUint(opt[5:], 10, 16)
if (err == nil)
{
port64, err := strconv.ParseUint(opt[5:], 10, 16)
if (err == nil)
{
activeNFS[int(port64)] = append(activeNFS[int(port64)], mount.MountPoint)
}
activeNFS[int(port64)] = append(activeNFS[int(port64)], mnt.MountPoint)
}
}
}
}
return activeNFS, nil
return activeNFS
}
// NodePublishVolume mounts the volume mounted to the staging path to the target path
@@ -1107,7 +1132,7 @@ func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandV
timeout := 30
var actualSize uint64
var err error
for i := 0; i < timeout && actualSize < requiredBytes; i++
for i := 0; i < timeout && actualSize < uint64(requiredBytes); i++
{
actualSize, err = GetBlockDeviceSize(devicePath)
if (err != nil)
@@ -1116,9 +1141,9 @@ func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandV
}
time.Sleep(1 * time.Second)
}
if (actualSize < requiredBytes)
if (actualSize < uint64(requiredBytes))
{
return nil, status.Error(codes.NotFound, "Failed to wait for online resize of %s to %v bytes: actual size is still %v after %v seconds",
return nil, status.Errorf(codes.NotFound, "Failed to wait for online resize of %s to %v bytes: actual size is still %v after %v seconds",
devicePath, requiredBytes, actualSize, timeout)
}
}

View File

@@ -0,0 +1,73 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
package vitastor
import (
"reflect"
"testing"
"k8s.io/utils/mount"
)
func TestCollectActiveNFS(t *testing.T)
{
mounts := []mount.MountInfo{
{
Root: "/",
Source: "127.0.0.1:/",
MountPoint: "/var/lib/kubelet/plugins/csi.vitastor.io/nfs/hash",
FsType: "nfs",
MountOptions: []string{"rw", "relatime"},
SuperOptions: []string{"rw", "vers=3", "port=45059"},
},
{
// This is how NFS subdirectory bind mounts appear on Linux 6.19:
// Root is still / and the subdirectory is only visible in Source.
Root: "/",
Source: "127.0.0.1:/pvc-one",
MountPoint: "/var/lib/kubelet/pods/pod-one/volumes/kubernetes.io~csi/pvc-one/mount",
FsType: "nfs",
MountOptions: []string{"rw", "relatime"},
SuperOptions: []string{"rw", "vers=3", "port=45059"},
},
{
Root: "/",
Source: "127.0.0.1:/pvc-two",
MountPoint: "/var/lib/kubelet/pods/pod-two/volumes/kubernetes.io~csi/pvc-two/mount",
FsType: "nfs",
MountOptions: []string{"rw", "relatime"},
SuperOptions: []string{"rw", "vers=3", "port=45059"},
},
{
// Older kernels expose the bind-mounted subdirectory in Root.
Root: "/pvc-three",
Source: "127.0.0.1:/",
MountPoint: "/var/lib/kubelet/pods/pod-three/volumes/kubernetes.io~csi/pvc-three/mount",
FsType: "nfs",
MountOptions: []string{"rw", "relatime"},
SuperOptions: []string{"rw", "vers=3", "port=45059"},
},
{
Root: "/",
Source: "server:/other",
MountPoint: "/mnt/other",
FsType: "nfs",
MountOptions: []string{"rw"},
SuperOptions: []string{"rw", "port=2049"},
},
}
got := collectActiveNFS(mounts)
want := map[int][]string{
45059: {
"/var/lib/kubelet/pods/pod-one/volumes/kubernetes.io~csi/pvc-one/mount",
"/var/lib/kubelet/pods/pod-two/volumes/kubernetes.io~csi/pvc-two/mount",
"/var/lib/kubelet/pods/pod-three/volumes/kubernetes.io~csi/pvc-three/mount",
},
}
if (!reflect.DeepEqual(got, want))
{
t.Fatalf("unexpected active NFS mounts: got %#v, want %#v", got, want)
}
}

2
debian/changelog vendored
View File

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

View File

@@ -1,4 +1,4 @@
VITASTOR_VERSION ?= v3.2.0
VITASTOR_VERSION ?= v3.2.1
all: build push

View File

@@ -4,7 +4,7 @@
#
# Desired Vitastor version
VITASTOR_VERSION=v3.2.0
VITASTOR_VERSION=v3.2.1
# Additional arguments for all containers
# For example, you may want to specify a custom logging driver here

View File

@@ -33,6 +33,7 @@ between clients, OSDs and etcd.
- [etcd_ws_keepalive_interval](#etcd_ws_keepalive_interval)
- [etcd_min_reload_interval](#etcd_min_reload_interval)
- [tcp_header_buffer_size](#tcp_header_buffer_size)
- [rxbounce](#rxbounce)
- [min_zerocopy_send_size](#min_zerocopy_send_size)
- [use_sync_send_recv](#use_sync_send_recv)
@@ -303,6 +304,21 @@ is received without an additional copy. You can try to play with this
parameter and see how it affects random iops and linear bandwidth if you
want.
## rxbounce
- Type: boolean
- Default: false
- Can be changed online: yes
Receive all TCP data into an intermediate buffer before copying to the actual
destination. Enabling this option is required for Windows guests if you use TCP
and [security.en.md#proto_checksums](proto_checksums) because Windows uses a
shared system-wide 'dummy page' to skip unneeded blocks during reads and breaks
checksum verification with it when data copy is avoided. rxbounce is only needed
when you use TCP and don't use transport encryption (AES-GCM); it's not required
with RDMA or with GCM encryption because both always copy data anyway (yes, RDMA
isn't actually zerocopy in Vitastor).
## min_zerocopy_send_size
- Type: integer

View File

@@ -33,6 +33,7 @@
- [etcd_ws_keepalive_interval](#etcd_ws_keepalive_interval)
- [etcd_min_reload_interval](#etcd_min_reload_interval)
- [tcp_header_buffer_size](#tcp_header_buffer_size)
- [rxbounce](#rxbounce)
- [min_zerocopy_send_size](#min_zerocopy_send_size)
- [use_sync_send_recv](#use_sync_send_recv)
@@ -310,6 +311,22 @@ Vitastor содержат 128-байтные заголовки, за котор
поменять этот параметр и посмотреть, как он влияет на производительность
случайного и линейного доступа.
## rxbounce
- Тип: булево (да/нет)
- Значение по умолчанию: false
- Можно менять на лету: да
Принимать все данные по TCP в промежуточный буфер и потом копировать их в
буфер клиента (ВМ). Это нужно для гостевых ОС Windows с TCP и включёнными
[security.ru.md#proto_checksums](proto_checksums), потому что Windows использует
одну разделяемую общесистемную страницу-заглушку для пропуска ненужных блоков
при чтении и ломает проверку контрольных сумм, если данные не копируются лишний
раз. Опция rxbounce нужна только если вы используете TCP и не используете
транспортное шифрование (AES-GCM); с RDMA или шифрованием опция не нужна, так
как в этом случае данные копируются всегда (да, RDMA в Vitastor на самом деле
не zero-copy).
## min_zerocopy_send_size
- Тип: целое число

View File

@@ -358,6 +358,29 @@
параметра читается без дополнительного копирования. Вы можете попробовать
поменять этот параметр и посмотреть, как он влияет на производительность
случайного и линейного доступа.
- name: rxbounce
type: bool
default: false
online: true
info: |
Receive all TCP data into an intermediate buffer before copying to the actual
destination. Enabling this option is required for Windows guests if you use TCP
and [security.en.md#proto_checksums](proto_checksums) because Windows uses a
shared system-wide 'dummy page' to skip unneeded blocks during reads and breaks
checksum verification with it when data copy is avoided. rxbounce is only needed
when you use TCP and don't use transport encryption (AES-GCM); it's not required
with RDMA or with GCM encryption because both always copy data anyway (yes, RDMA
isn't actually zerocopy in Vitastor).
info_ru: |
Принимать все данные по TCP в промежуточный буфер и потом копировать их в
буфер клиента (ВМ). Это нужно для гостевых ОС Windows с TCP и включёнными
[security.ru.md#proto_checksums](proto_checksums), потому что Windows использует
одну разделяемую общесистемную страницу-заглушку для пропуска ненужных блоков
при чтении и ломает проверку контрольных сумм, если данные не копируются лишний
раз. Опция rxbounce нужна только если вы используете TCP и не используете
транспортное шифрование (AES-GCM); с RDMA или шифрованием опция не нужна, так
как в этом случае данные копируются всегда (да, RDMA в Vitastor на самом деле
не zero-copy).
- name: min_zerocopy_send_size
type: int
default: 32768

View File

@@ -26,9 +26,9 @@ at Vitastor Kubernetes operator: https://github.com/Antilles7227/vitastor-operat
The instruction is very simple.
1. Download a Docker image of the desired version: \
`docker pull vitalif/vitastor:v3.2.0`
`docker pull vitalif/vitastor:v3.2.1`
2. Install scripts to the host system: \
`docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.0 install.sh`
`docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
3. Reload udev rules: \
`udevadm control --reload-rules`
4. Enable the vitastor-host service: \

View File

@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
Инструкция по установке максимально простая.
1. Скачайте Docker-образ желаемой версии: \
`docker pull vitalif/vitastor:v3.2.0`
`docker pull vitalif/vitastor:v3.2.1`
2. Установите скрипты в хост-систему командой: \
`docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.0 install.sh`
`docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
3. Перезагрузите правила udev: \
`udevadm control --reload-rules`
4. Включите сервис vitastor-host: \

View File

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

View File

@@ -1,6 +1,6 @@
{
"name": "vitastor",
"version": "3.2.0",
"version": "3.2.1",
"description": "Low-level native bindings to Vitastor client library",
"main": "index.js",
"keywords": [

View File

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

View File

@@ -0,0 +1,699 @@
From 560382fdf86fdbbd08c38187ef602c04e0abece0 Mon Sep 17 00:00:00 2001
From: Vitaliy Filippov <vitalif@yourcmc.ru>
Date: Sat, 5 Sep 2026 20:41:38 +0300
Subject: [PATCH] Add Vitastor support
---
include/libvirt/libvirt-storage.h | 1 +
src/conf/domain_conf.c | 4 +-
src/conf/domain_validate.c | 11 +-
src/conf/schemas/domaincommon.rng | 30 +++++
src/conf/storage_conf.c | 20 ++-
src/conf/storage_conf.h | 2 +
src/conf/storage_source_conf.c | 2 +
src/conf/storage_source_conf.h | 1 +
src/conf/virstorageobj.c | 3 +
src/libvirt-storage.c | 1 +
src/libxl/libxl_conf.c | 1 +
src/libxl/xen_xl.c | 1 +
src/qemu/qemu_block.c | 45 +++++++
src/qemu/qemu_domain.c | 5 +-
src/qemu/qemu_nbdkit.c | 2 +
src/qemu/qemu_snapshot.c | 2 +
src/storage/storage_driver.c | 1 +
.../storage_source_backingstore.c | 123 ++++++++++++++++++
src/test/test_driver.c | 1 +
.../storagepoolcapsschemadata/poolcaps-fs.xml | 7 +
.../poolcaps-full.xml | 7 +
tests/storagepoolxml2argvtest.c | 1 +
tools/virsh-pool.c | 3 +
23 files changed, 268 insertions(+), 6 deletions(-)
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 541dad5bdc..5cb94e1fed 100644
--- a/src/conf/domain_conf.c
+++ b/src/conf/domain_conf.c
@@ -7453,7 +7453,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)
@@ -32196,6 +32197,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 4558e7b210..3c6b2a8ee8 100644
--- a/src/conf/domain_validate.c
+++ b/src/conf/domain_validate.c
@@ -520,6 +520,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:
@@ -592,7 +593,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) {
@@ -600,10 +601,14 @@ 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 1f9ac102a0..51b55ea72b 100644
--- a/src/conf/schemas/domaincommon.rng
+++ b/src/conf/schemas/domaincommon.rng
@@ -2092,6 +2092,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">
@@ -2442,6 +2471,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 1dc9365bf2..a8a736be81 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 087de1eaf2..acc046816b 100644
--- a/src/conf/storage_source_conf.c
+++ b/src/conf/storage_source_conf.c
@@ -89,6 +89,7 @@ VIR_ENUM_IMPL(virStorageNetProtocol,
"ssh",
"vxhs",
"nfs",
+ "vitastor",
);
@@ -1313,6 +1314,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 fc868b31af..1c9fbc02b9 100644
--- a/src/conf/storage_source_conf.h
+++ b/src/conf/storage_source_conf.h
@@ -130,6 +130,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 2b988157fa..9d0eb47b25 100644
--- a/src/libxl/libxl_conf.c
+++ b/src/libxl/libxl_conf.c
@@ -1069,6 +1069,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 b2ff0edcf2..5d9f50c88c 100644
--- a/src/libxl/xen_xl.c
+++ b/src/libxl/xen_xl.c
@@ -1461,6 +1461,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 a7062d3e96..4f2b6cc0e6 100644
--- a/src/qemu/qemu_block.c
+++ b/src/qemu/qemu_block.c
@@ -743,6 +743,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 *
qemuBlockStorageSourceGetSshProps(virStorageSource *src)
{
@@ -1093,6 +1125,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_SSH:
driver = "ssh";
if (!(fileprops = qemuBlockStorageSourceGetSshProps(src)))
@@ -1996,6 +2034,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:
@@ -2376,6 +2415,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_SSH:
if (srcPriv->nbdkitProcess) {
/* disk creation not yet supported with nbdkit, and even if it
diff --git a/src/qemu/qemu_domain.c b/src/qemu/qemu_domain.c
index ac56fc7cb4..b80b311eea 100644
--- a/src/qemu/qemu_domain.c
+++ b/src/qemu/qemu_domain.c
@@ -4677,7 +4677,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;
@@ -4693,6 +4694,7 @@ qemuDomainValidateStorageSource(virStorageSource *src,
case VIR_STORAGE_NET_PROTOCOL_ISCSI:
case VIR_STORAGE_NET_PROTOCOL_NBD:
case VIR_STORAGE_NET_PROTOCOL_RBD:
+ case VIR_STORAGE_NET_PROTOCOL_VITASTOR:
case VIR_STORAGE_NET_PROTOCOL_SSH:
case VIR_STORAGE_NET_PROTOCOL_SHEEPDOG:
break;
@@ -9103,6 +9105,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_nbdkit.c b/src/qemu/qemu_nbdkit.c
index c1bc6bc363..9d3433bf75 100644
--- a/src/qemu/qemu_nbdkit.c
+++ b/src/qemu/qemu_nbdkit.c
@@ -869,6 +869,7 @@ qemuNbdkitInitStorageSource(qemuNbdkitCaps *caps WITHOUT_NBDKIT_UNUSED,
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:
@@ -1133,6 +1134,7 @@ qemuNbdkitProcessBuildCommand(qemuNbdkitProcess *proc)
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/qemu/qemu_snapshot.c b/src/qemu/qemu_snapshot.c
index 302775af92..9d8fa0ded2 100644
--- a/src/qemu/qemu_snapshot.c
+++ b/src/qemu/qemu_snapshot.c
@@ -665,6 +665,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:
@@ -890,6 +891,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 e19e032427..59f91f4710 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 821378883c..2211f6891b 100644
--- a/src/storage_file/storage_source_backingstore.c
+++ b/src/storage_file/storage_source_backingstore.c
@@ -264,6 +264,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)
@@ -379,6 +448,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:
@@ -953,6 +1027,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,
@@ -1130,6 +1252,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 1165689de7..bba846351c 100644
--- a/src/test/test_driver.c
+++ b/src/test/test_driver.c
@@ -7345,6 +7345,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 d5c2531ab8..b19308ac38 100644
--- a/tests/storagepoolxml2argvtest.c
+++ b/tests/storagepoolxml2argvtest.c
@@ -57,6 +57,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 2010ef1356..072e2ff9e8 100644
--- a/tools/virsh-pool.c
+++ b/tools/virsh-pool.c
@@ -1187,6 +1187,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;
}
--
2.51.0

View File

@@ -0,0 +1,91 @@
--- qemu-kvm.spec.orig 2026-08-24 00:00:00.000000000 +0000
+++ qemu-kvm.spec 2026-09-05 16:48:22.226624885 +0000
@@ -102,8 +102,6 @@
%global target_list %{kvm_target}-softmmu
-%global block_drivers_rw_list qcow2,raw,file,host_device,nbd,iscsi,rbd,blkdebug,luks,null-co,nvme,copy-on-read,throttle,compress,virtio-blk-vhost-vdpa,virtio-blk-vfio-pci,virtio-blk-vhost-user,io_uring,nvme-io_uring
-%global block_drivers_ro_list vdi,vmdk,vhdx,vpc,https
%define qemudocdir %{_docdir}/%{name}
%global firmwaredirs "%{_datadir}/qemu-firmware:%{_datadir}/ipxe/qemu:%{_datadir}/seavgabios:%{_datadir}/seabios"
@@ -134,6 +132,7 @@ Requires: %{name}-block-blkio = %{epoch}
%if %{have_block_rbd} \
Requires: %{name}-block-rbd = %{epoch}:%{version}-%{release} \
%endif \
+Requires: %{name}-block-vitastor = %{epoch}:%{version}-%{release}\
Requires: %{name}-audio-pa = %{epoch}:%{version}-%{release}
# Since SPICE is removed from RHEL-9, the following Obsoletes:
@@ -158,7 +157,7 @@ Obsoletes: %{name}-block-ssh <= %{epoch}
Summary: QEMU is a machine emulator and virtualizer
Name: qemu-kvm
Version: 10.1.0
-Release: 16%{?rcrel}%{?dist}%{?cc_suffix}.5.alma.1
+Release: 16.vitastor%{?rcrel}%{?dist}%{?cc_suffix}.5.alma.1
# Epoch because we pushed a qemu-1.0 package. AIUI this can't ever be dropped
# Epoch 15 used for RHEL 8
# Epoch 17 used for RHEL 9 (due to release versioning offset in RHEL 8.5)
@@ -181,6 +180,7 @@ Source28: 95-kvm-memlock.conf
Source30: kvm-s390x.conf
Source31: kvm-x86.conf
Source36: README.tests
+Source37: qemu-vitastor.c
Patch0004: 0004-Initial-redhat-build.patch
@@ -471,6 +471,7 @@ Patch149: kvm-file-posix-populate-pwrite
Patch150: kvm-block-use-pwrite_zeroes_alignment-when-writing-first.patch
# For RHEL-207389 - qemu-img create/convert fails on target block device with 4k sector size [rhel-10.2.z]
Patch151: kvm-iotests-add-Linux-loop-device-image-creation-test.patch
+Patch152: qemu-10.1-vitastor.patch
# AlmaLinux Patch
Patch2001: 2001-Add-ppc64-support.patch
@@ -509,6 +510,7 @@ BuildRequires: libcurl-devel
%if %{have_block_rbd}
BuildRequires: librbd-devel
%endif
+BuildRequires: vitastor-client-devel
# We need both because the 'stap' binary is probed for by configure
BuildRequires: systemtap
BuildRequires: systemtap-sdt-devel
@@ -729,6 +731,13 @@ using the rbd protocol.
%endif
+%package block-vitastor
+Summary: QEMU Vitastor block driver
+Requires: %{name}-common%{?_isa} = %{epoch}:%{version}-%{release}
+%description block-vitastor
+This package provides the additional Vitastor block driver for QEMU.
+
+
%package audio-pa
Summary: QEMU PulseAudio audio driver
Requires: %{name}-common%{?_isa} = %{epoch}:%{version}-%{release}
@@ -831,6 +840,7 @@ This package provides the additional D-B
%prep
%setup -q -n qemu-%{version}%{?rcstr}
%autopatch -p1
+cp %{SOURCE37} ./block/vitastor.c
%global qemu_kvm_build qemu_kvm_build
mkdir -p %{qemu_kvm_build}
@@ -1067,6 +1077,7 @@ run_configure \
%if %{have_block_rbd}
--enable-rbd \
%endif
+ --enable-vitastor \
%if %{have_librdma}
--enable-rdma \
%endif
@@ -1564,6 +1575,8 @@ useradd -r -u 107 -g qemu -G kvm -d / -s
%files block-rbd
%{_libdir}/%{name}/block-rbd.so
%endif
+%files block-vitastor
+%{_libdir}/%{name}/block-vitastor.so
%files audio-pa
%{_libdir}/%{name}/audio-pa.so

View File

@@ -1,11 +1,11 @@
Name: vitastor
Version: 3.2.0
Version: 3.2.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-3.2.0.el10.tar.gz
Source0: vitastor-3.2.1.el10.tar.gz
BuildRequires: gperftools-devel
BuildRequires: gcc-c++

View File

@@ -1,11 +1,11 @@
Name: vitastor
Version: 3.2.0
Version: 3.2.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-3.2.0.el7.tar.gz
Source0: vitastor-3.2.1.el7.tar.gz
BuildRequires: gperftools-devel
BuildRequires: devtoolset-9-gcc-c++

View File

@@ -1,11 +1,11 @@
Name: vitastor
Version: 3.2.0
Version: 3.2.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-3.2.0.el8.tar.gz
Source0: vitastor-3.2.1.el8.tar.gz
BuildRequires: gperftools-devel
BuildRequires: gcc-toolset-9-gcc-c++

View File

@@ -1,11 +1,11 @@
Name: vitastor
Version: 3.2.0
Version: 3.2.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-3.2.0.el9.tar.gz
Source0: vitastor-3.2.1.el9.tar.gz
BuildRequires: gperftools-devel
BuildRequires: gcc-c++

View File

@@ -21,7 +21,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
endif()
set(ENABLE_COVERAGE false CACHE BOOL "Enable code coverage")
add_definitions(-DVITASTOR_VERSION="3.2.0")
add_definitions(-DVITASTOR_VERSION="3.2.1")
add_definitions(-D_GNU_SOURCE -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 -fvisibility=hidden -I ${CMAKE_SOURCE_DIR}/src)
add_link_options(-fno-omit-frame-pointer)
if (${WITH_ASAN})

View File

@@ -80,6 +80,16 @@ int journal_flusher_t::get_syncing_buffer()
return syncing_buffer;
}
uint64_t journal_flusher_t::get_flushing_lsn(object_id oid)
{
auto it = flushing.find(oid);
if (it == flushing.end())
{
return 0;
}
return it->second;
}
bool journal_flusher_t::is_active()
{
return active_flushers > 0;
@@ -250,7 +260,6 @@ resume_16:
}
assert(false);
}
flusher->flushing.insert(cur_oid);
resume_1:
wait_state = 1;
should_repeat = false;
@@ -258,7 +267,6 @@ resume_1:
if (!cur_obj)
{
// Object does not exist
flusher->flushing.erase(cur_oid);
goto resume_0;
}
// Scan versions to flush
@@ -291,10 +299,10 @@ resume_1:
if (!compact_info.compact_lsn)
{
// Flushing is aborted
flusher->flushing.erase(cur_oid);
bs->heap->unlock_entry(cur_oid);
goto resume_0;
}
flusher->flushing[cur_oid] = compact_info.compact_lsn;
flusher->active_flushers++;
if (!compact_info.do_delete && compact_info.clean_wr->type() == BS_HEAP_BIG_INTENT &&
(!bs->dsk.csum_block_size || bs->dsk.csum_block_size > bs->dsk.bitmap_granularity))
@@ -519,12 +527,12 @@ resume_13:
printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_info.compact_lsn, copy_count);
}
flusher->active_flushers--;
flusher->flushing.erase(cur_oid);
if (should_repeat)
{
// Flush the same object again
goto resume_1;
}
flusher->flushing.erase(cur_oid);
// All done
goto resume_0;
}

View File

@@ -99,7 +99,7 @@ class journal_flusher_t
blockstore_impl_t *bs;
friend class journal_flusher_co;
robin_hood::unordered_flat_set<object_id> flushing;
robin_hood::unordered_flat_map<object_id, uint64_t> flushing;
int active_flushers = 0;
std::list<flusher_data_sync_t> data_syncs;
int wanting_meta_fsync = 0;
@@ -112,6 +112,7 @@ public:
~journal_flusher_t();
void loop();
int get_syncing_buffer();
uint64_t get_flushing_lsn(object_id oid);
bool is_active();
void request_trim();
void release_trim();

View File

@@ -919,6 +919,19 @@ void blockstore_heap_t::recheck_full_gc()
}
}
// The entries which were hiding under the ones just dropped are the newest ones of the object
// now, and their data may be unfinished just as well - a redirect intent acked without a data
// fsync, or a buffered small write, both of which the outage may have left without their data.
// So look at what became visible and recheck it too
void blockstore_heap_t::recheck_requeue(object_id oid)
{
auto obj = read_entry(oid);
if (obj)
{
recheck_queue.push_back(obj);
}
}
void blockstore_heap_t::recheck_drop_entries(heap_entry_t *obj, heap_entry_t *bad_wr)
{
// write entry is invalid, erase it and all newer entries
@@ -964,13 +977,19 @@ void blockstore_heap_t::recheck_drop_entries(heap_entry_t *obj, heap_entry_t *ba
// same object overwrite. Compaction merges those newer writes into the data block in place: it
// changes bytes which <wr>'s checksums still describe, without touching <wr> itself. Nothing ever
// reads those bytes from <wr> - they're shadowed - so a mismatch there means nothing.
// A shadow only holds while the entry casting it stays, though: the very same recheck may find
// that newer entry unfinished and drop it, and then the bytes under it become visible again.
// So entries which this recheck is about to drop cast no shadow. This is why the entry needing
// a data read is always verified last, see recheck_start_reads() - by then it is already known
// which of the entries above it survive.
// This is only exact while a checksum block isn't larger than the write granularity. With larger
// blocks an overwritten part poisons the checksum of the whole block, and skipping the block
// would hide a genuinely unfinished write in the rest of it - so there the flusher instead makes
// sure such an entry is below the persisted completed_lsn and never gets rechecked at all,
// see the big_intent trim in journal_flusher_co::loop()
bool blockstore_heap_t::recheck_verify(heap_entry_t *obj, heap_entry_t *wr, uint8_t *buf)
bool blockstore_heap_t::recheck_verify(heap_recheck_state_t *st, heap_entry_t *wr, uint8_t *buf)
{
auto obj = st->obj;
if (!dsk->csum_block_size || dsk->csum_block_size > dsk->bitmap_granularity ||
wr == obj || wr->type() == BS_HEAP_SMALL_WRITE)
{
@@ -1002,7 +1021,8 @@ bool blockstore_heap_t::recheck_verify(heap_entry_t *obj, heap_entry_t *wr, uint
bool shadowed = false;
for (auto newer = obj; newer && newer != wr && !shadowed; newer = prev(newer))
{
if (newer->type() != BS_HEAP_SMALL_WRITE && newer->type() != BS_HEAP_INTENT_WRITE)
if (newer->type() != BS_HEAP_SMALL_WRITE && newer->type() != BS_HEAP_INTENT_WRITE ||
st->bad_wr && newer->lsn >= st->bad_wr->lsn)
continue;
shadowed = newer->small().offset < pos+dsk->csum_block_size &&
newer->small().offset+newer->small().len > pos;
@@ -1015,14 +1035,14 @@ bool blockstore_heap_t::recheck_verify(heap_entry_t *obj, heap_entry_t *wr, uint
return ok;
}
void blockstore_heap_t::recheck_start_reads(heap_recheck_state_t *st)
int blockstore_heap_t::recheck_start_reads(heap_recheck_state_t *st)
{
int started = 0;
if (st->sent_reads >= st->total_reads)
return;
return 0;
while (recheck_in_progress < recheck_queue_depth)
{
auto wr = skip_commits(st->next_wr);
st->next_wr = prev(wr);
uint64_t loc = 0, len = 0;
bool from_data = false;
if (wr->type() == BS_HEAP_SMALL_WRITE)
@@ -1057,6 +1077,15 @@ void blockstore_heap_t::recheck_start_reads(heap_recheck_state_t *st)
len = wr->small().len;
from_data = true;
}
if (from_data && st->checked_reads < st->sent_reads)
{
// Part of this entry's data may be shadowed by the newer entries of the same
// object, and whether it really is depends on whether they survive the recheck.
// So verify it only when all of them are already verified - see recheck_verify()
break;
}
st->next_wr = prev(wr);
st->oldest_wr = wr;
st->sent_reads++;
recheck_in_progress++;
recheck_pending_reads--;
@@ -1064,20 +1093,32 @@ void blockstore_heap_t::recheck_start_reads(heap_recheck_state_t *st)
recheck_cb(from_data, loc, len, [this, st, wr](uint8_t *buf)
{
st->checked_reads++;
if (!recheck_verify(st->obj, wr, buf))
if (!recheck_verify(st, wr, buf))
st->bad_wr = !st->bad_wr || st->bad_wr->lsn > wr->lsn ? wr : st->bad_wr;
if (st->checked_reads >= st->total_reads)
{
if (st->bad_wr)
recheck_drop_entries(st->obj, st->bad_wr);
recheck_states.erase(st->obj);
auto bad_wr = st->bad_wr;
auto obj = st->obj;
// Entries below the oldest one we read are unverified, the rest of the set
// is already known to be good
bool check_more = (bad_wr && bad_wr == st->oldest_wr);
recheck_states.erase(obj);
if (bad_wr)
{
object_id oid = { .inode = obj->inode, .stripe = obj->stripe };
recheck_drop_entries(obj, bad_wr);
if (check_more)
recheck_requeue(oid);
}
}
recheck_in_progress--;
recheck_small_writes(NULL, 0);
});
started++;
if (is_last)
break;
}
return started;
}
bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, uint64_t offset, uint64_t len, std::function<void(uint8_t* buf)>)> read_buffer, int queue_depth)
@@ -1101,8 +1142,14 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
in_recheck = true;
while (recheck_pending_reads > 0 && recheck_in_progress < recheck_queue_depth)
{
int started = 0;
for (auto & sp: recheck_states)
recheck_start_reads(&sp.second);
started += recheck_start_reads(&sp.second);
if (!started)
{
// Everything left is postponed until the reads in flight complete
break;
}
}
while (recheck_queue.size() > 0 && recheck_in_progress < recheck_queue_depth)
{
@@ -1110,20 +1157,33 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
recheck_queue.pop_front();
bool need_data = false;
int recheck_count = count_recheck_entries(obj, need_data);
if (!recheck_count)
{
// Nothing left to recheck - the object was requeued after a drop which exposed
// entries that are already known to be good
continue;
}
if (!need_data && buffer_area)
{
// Check this object synchronously
heap_entry_t *bad_wr = NULL;
heap_entry_t *bad_wr = NULL, *oldest_wr = NULL;
auto wr = skip_commits(obj);
for (int i = 0; i < recheck_count; i++, wr = skip_commits(prev(wr)))
{
fprintf(stderr, "Notice: rechecking %jx:%jx l%ju - %u bytes at %ju in buffer area\n",
wr->inode, wr->stripe, wr->lsn, wr->small().len, wr->small().location);
oldest_wr = wr;
if (!calc_checksums(wr, buffer_area + wr->small().location, false))
bad_wr = wr;
}
if (bad_wr)
{
object_id oid = { .inode = obj->inode, .stripe = obj->stripe };
bool check_more = (bad_wr == oldest_wr);
recheck_drop_entries(obj, bad_wr);
if (check_more)
recheck_requeue(oid);
}
}
else
{
@@ -2337,10 +2397,11 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
{
if (wr->type() == BS_HEAP_ROLLBACK && wr->lsn <= fsynced_lsn)
{
// compact_version in commit/rollback is taken from actual write entries
// to handle lsn=0 reset correctly
if (!res.compact_lsn)
{
res.compact_lsn = wr->lsn;
res.compact_version = wr->version;
}
if (rollback_version > wr->version)
{
@@ -2350,14 +2411,15 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
}
if (wr->type() == BS_HEAP_COMMIT && wr->lsn <= fsynced_lsn)
{
res.do_delete = false;
if (!res.compact_lsn)
{
res.compact_lsn = wr->lsn;
res.compact_version = wr->version;
}
res.do_delete = false;
if (commit_version < wr->version)
{
commit_version = wr->version;
}
continue;
}
bool rolled_back = (wr->version > rollback_version);
@@ -2370,6 +2432,7 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
if (!stable && !committed || wr->lsn > fsynced_lsn)
{
// Unstable and non-fsynced writes can't be compacted yet
// Reset newer compact_lsn to place compact entry correctly
res.do_delete = false;
res.compact_lsn = 0;
res.compact_version = 0;
@@ -2386,6 +2449,9 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
if (!stable && !res.compact_lsn)
{
res.compact_lsn = wr->lsn;
}
if (res.compact_version < wr->version)
{
res.compact_version = wr->version;
}
res.clean_wr = wr;
@@ -2402,6 +2468,9 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
if (!res.compact_lsn)
{
res.compact_lsn = wr->lsn;
}
if (res.compact_version < wr->version)
{
res.compact_version = wr->version;
}
res.do_delete = false;

View File

@@ -167,6 +167,9 @@ struct heap_recheck_state_t
size_t sent_reads = 0;
size_t checked_reads = 0;
heap_entry_t *bad_wr = NULL;
// The oldest entry a read was issued for. Everything below it is unverified, so if it
// turns out to be bad the object has to be looked at again after dropping it
heap_entry_t *oldest_wr = NULL;
};
using i64hash_t = robin_hood::hash<uint64_t>;
@@ -233,8 +236,9 @@ class blockstore_heap_t
uint64_t get_pg_id(inode_t inode, uint64_t stripe);
bool validate_object(heap_entry_t *obj);
void fill_recheck_queue();
void recheck_requeue(object_id oid);
void recheck_drop_entries(heap_entry_t *obj, heap_entry_t *bad_wr);
void recheck_start_reads(heap_recheck_state_t *st);
int recheck_start_reads(heap_recheck_state_t *st);
int mark_used_blocks();
void init_free_bad_entry(heap_entry_t *wr);
void init_erase_bad_entry(heap_list_item_t *li);
@@ -259,7 +263,7 @@ class blockstore_heap_t
void mark_completed_lsns(uint64_t mod_lsn);
heap_entry_t *skip_commits(heap_entry_t *wr);
int count_recheck_entries(heap_entry_t *obj, bool & need_data);
bool recheck_verify(heap_entry_t *obj, heap_entry_t *wr, uint8_t *buf);
bool recheck_verify(heap_recheck_state_t *st, heap_entry_t *wr, uint8_t *buf);
void apply_inflight(heap_inflight_lsn_t & inflight);
public:
blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0);

View File

@@ -77,34 +77,36 @@ bool blockstore_impl_t::intent_write_allowed(blockstore_op_t *op, heap_entry_t *
{
return false;
}
bool ok = true;
bool ok = false;
heap->iterate_with_stable(obj, obj->lsn, [&](heap_entry_t *wr, bool stable)
{
// Intent writes are not allowed over buffered writes
auto t = wr->type();
if (t == BS_HEAP_SMALL_WRITE)
{
ok = false;
return false;
}
// Intent writes are not allowed over unstable writes
if (!stable)
{
ok = false;
// Intent writes are not allowed over unstable writes
return false;
}
// Intent writes are not allowed over unfinished intent writes
if ((t == BS_HEAP_INTENT_WRITE || t == BS_HEAP_BIG_INTENT) && wr->lsn > heap->get_fsynced_lsn())
else if (t == BS_HEAP_INTENT_WRITE || t == BS_HEAP_BIG_INTENT)
{
ok = false;
return false;
// Intent writes are not allowed over unfinished intent writes
ok = (wr->lsn <= heap->get_fsynced_lsn());
}
// Intent writes are allowed over BIG_WRITEs even with fsyncs because BIG_WRITE is always counted as fsynced
if (t == BS_HEAP_BIG_WRITE || t == BS_HEAP_BIG_INTENT)
// Intent writes are allowed over BIG_WRITEs even with enable data_fsync
// because BIG_WRITEs are always fsynced before adding the entry
else if (t == BS_HEAP_BIG_WRITE)
{
return false;
// But sadly with one exception: a BIG_WRITE may be a result of compaction, with a
// reused older LSN, still in progress of being written. In that case it can still
// disappear after a reboot, and if it disappears, an older small_write may be replayed
// again over our intent write data if it's there. So we have to check the previous write too...
if (wr->lsn == flusher->get_flushing_lsn({ .inode = wr->inode, .stripe = wr->stripe }))
{
return true;
}
ok = true;
}
return true;
// Intent writes are not allowed over buffered writes and other write types
return false;
});
return ok;
}

View File

@@ -223,8 +223,11 @@ void blockstore_impl_t::loop()
// its own state, so a second sync started while the first one is still in flight
// would find nothing left to sync and report success without anything having
// reached the disk - while the writes it was supposed to cover are still in the
// first sync's hands
if (!has_unfinished_sync)
// first sync's hands.
// A deletion takes them over in the very same way, see
// forget_unstable_before_delete() - so wait for it to be journaled and acked,
// and it lands in the unsynced lists itself by then
if (!has_unfinished_sync && !unsynced_forgotten)
{
wr_st = continue_sync(op);
has_unfinished_sync = (wr_st != 2);
@@ -425,6 +428,7 @@ void blockstore_impl_t::init_op(blockstore_op_t *op)
PRIV(op)->wait_for = 0;
PRIV(op)->op_state = 0;
PRIV(op)->pending_ops = 0;
PRIV(op)->forgot_unsynced = false;
}
static bool replace_stable(object_id oid, uint64_t version, int search_start, int search_end, obj_ver_id* list)

View File

@@ -111,6 +111,9 @@ struct blockstore_op_private_t
// Warning: must not have a default value here because it's written to before calling constructor in blockstore_write.cpp O_o
uint64_t real_version;
timespec tv_begin;
// Set on a deletion which dropped preceding unsynced writes of the same object,
// see forget_unstable_before_delete()
bool forgot_unsynced;
// Sync
std::vector<obj_ver_id> sync_big_writes, sync_small_writes;
@@ -171,6 +174,9 @@ class blockstore_impl_t: public blockstore_i
std::set<blockstore_op_t*> own_ops;
std::vector<obj_ver_id> unsynced_big_writes, unsynced_small_writes;
int unsynced_big_write_count = 0, unstable_unsynced = 0;
// Number of deletions which took over the durability of preceding unsynced writes but
// aren't journaled themselves yet, see forget_unstable_before_delete()
int unsynced_forgotten = 0;
int unsynced_queued_ops = 0;
allocator_t *data_alloc = NULL;
uint64_t used_blocks = 0;

View File

@@ -28,7 +28,7 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
else
{
dirty_it--;
if (dirty_it->first.oid != ov.oid || dirty_it->first.version < ov.version)
if (dirty_it->first.oid != ov.oid || dirty_it->first.version <= ov.version)
{
// Already rolled back, skip this object version
return STAB_SPLIT_DONE;
@@ -40,8 +40,12 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
// Object write is still in progress. Wait until the write request completes
return STAB_SPLIT_WAIT;
}
else if (!IS_SYNCED(dirty_it->second.state) ||
IS_STABLE(dirty_it->second.state))
else if (IS_STABLE(dirty_it->second.state))
{
// Can't jump back over a stable version
return -EBUSY;
}
else if (!IS_SYNCED(dirty_it->second.state))
{
// Sync the object
return STAB_SPLIT_SYNC;

View File

@@ -64,6 +64,11 @@ bool blockstore_impl_t::forget_unstable_before_delete(blockstore_op_t *op, block
}
if (first_it == dirty_it)
return true;
// Their durability becomes ours: they're acknowledged, but from now on the only thing
// on the disk which accounts for them is our deletion. A sync started before we're
// journaled and acked would otherwise find nothing to do and report success
unsynced_forgotten++;
PRIV(op)->forgot_unsynced = true;
for (auto it = first_it; it != dirty_it; it++)
{
// Drop it from the unsynced lists so that a later sync doesn't journal it
@@ -129,6 +134,10 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
if (op->opcode == BS_OP_WRITE_STABLE && !IS_INSTANT(dirty_it->second.state))
{
op->retval = -EINVAL;
if (!is_del && alloc_dyn_data)
{
free(dyn);
}
return false;
}
found = true;
@@ -331,6 +340,11 @@ void blockstore_impl_t::unshift_stable_flush(obj_ver_id ov)
void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, blockstore_dirty_db_t::iterator dirty_it, int retval)
{
if (PRIV(op)->forgot_unsynced)
{
PRIV(op)->forgot_unsynced = false;
unsynced_forgotten--;
}
while (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
{
free_dirty_dyn_data(dirty_it->second);
@@ -799,6 +813,12 @@ resume_4:
});
}
}
if (PRIV(op)->forgot_unsynced)
{
// We're in the unsynced lists ourselves now, so a sync may proceed again
PRIV(op)->forgot_unsynced = false;
unsynced_forgotten--;
}
if (imm && (dirty_it->second.state & BS_ST_TYPE_MASK) == BS_ST_BIG_WRITE)
{
// Unblock small writes

View File

@@ -10,6 +10,7 @@
#include <stdexcept>
#include "addr_util.h"
#include "json_util.h"
#include "str_util.h"
#include "messenger.h"
#ifdef WITH_RDMA
@@ -244,6 +245,7 @@ void osd_messenger_t::parse_config(const json11::Json & config, bool init)
if (!this->osd_ping_timeout)
this->osd_ping_timeout = 5;
this->log_level = config["log_level"].uint64_value();
this->rxbounce = json_is_true(config["rxbounce"]);
// All other parameters are only set on init
if (!init)
return;
@@ -251,13 +253,13 @@ void osd_messenger_t::parse_config(const json11::Json & config, bool init)
if (!config["use_rdma"].is_null())
{
// RDMA is on by default in RDMA-enabled builds
this->use_rdma = config["use_rdma"].bool_value() || config["use_rdma"].uint64_value() != 0;
this->use_rdma = json_is_true(config["use_rdma"].bool_value());
}
#ifdef WITH_RDMACM
// Use RDMA CM? (required for iWARP and may be useful for IB)
// FIXME: Only parse during start
this->use_rdmacm = config["use_rdmacm"].bool_value() || config["use_rdmacm"].uint64_value() != 0;
this->disable_tcp = this->use_rdmacm && (config["disable_tcp"].bool_value() || config["disable_tcp"].uint64_value() != 0);
this->use_rdmacm = json_is_true(config["use_rdmacm"]);
this->disable_tcp = this->use_rdmacm && json_is_true(config["disable_tcp"]);
#endif
this->rdma_device = config["rdma_device"].string_value();
this->rdma_port_num = (uint8_t)config["rdma_port_num"].uint64_value();

View File

@@ -53,10 +53,17 @@
#define MAX_SIMPLE_PAYLOAD_SIZE 1048576
#define MSGR_SENDP_BUF 1
#define MSGR_SENDP_IN 2
#define MSGR_SENDP_OUT 4
#define MSGR_SENDP_REF 8
struct osd_client_t;
struct msgr_sendp_t
{
osd_op_t *op;
int flags;
void *buf;
uint32_t flags;
};
#ifdef WITH_RDMA
@@ -151,13 +158,14 @@ struct osd_client_t
bool write_ready = true;
std::vector<iovec> send_list;
size_t send_list_size = 0;
std::deque<osd_op_t*> send_free_ops;
std::vector<osd_op_t*> zc_free_list;
uint64_t next_send_free = 0;
std::deque<msgr_sendp_t> send_free;
op_aes_xts_encrypt_t *xts_enc_ctx = NULL;
XXH3_state_t* write_csum_state = NULL;
~osd_client_t();
void cancel_ops();
void unref_send_free(uint64_t from, uint64_t to);
};
struct osd_wanted_peer_t
@@ -222,6 +230,7 @@ protected:
int min_zerocopy_send_size = DEFAULT_MIN_ZEROCOPY_SEND_SIZE;
int iothread_count = 0;
int max_cipher_pool_size = 256;
bool rxbounce = false;
std::string client_cert;
std::string client_key;
@@ -346,8 +355,7 @@ protected:
bool try_send(osd_client_t *cl);
bool wakeup_send(osd_client_t *cl);
bool handle_send(int result, bool prev, bool more, osd_client_t *cl);
static void post_send_free(osd_op_t *op);
bool handle_send(int result, bool prev, bool more, osd_client_t *cl, uint64_t free_from, uint64_t free_to);
bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr);
void next_write_op(osd_client_t *cl);
bool op_write_buf(osd_client_t *cl, uint8_t *src, size_t src_len, uint8_t *dst, size_t dst_len, bool skip_csum, size_t & from, size_t & done);

View File

@@ -181,7 +181,7 @@ struct __attribute__((visibility("default"))) osd_op_t
void *buf = NULL;
// bitmap, bitmap_len, bmp_data are only meaningful for reads
void *bitmap = NULL;
unsigned bitmap_len = 0;
uint32_t bitmap_len = 0;
size_t bmp_data = 0;
uint8_t *bitmap_buf = NULL;
void *rmw_buf = NULL;

View File

@@ -598,6 +598,7 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
: rc->send_done_pos-rc->send_out_pos);
if (dst_len > rc->max_msg)
dst_len = rc->max_msg;
uint64_t send_free_from = cl->next_send_free + cl->send_free.size();
copied = copy_ops_to(cl, dst, dst_len);
if (cl->io_error)
{
@@ -619,7 +620,7 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
};
try_send_rdma_wr(cl, &sge, 1);
rc->send_sizes.push_back(copied);
cl->send_free_ops.push_back(NULL); // end marker
cl->unref_send_free(send_free_from, cl->next_send_free + cl->send_free.size()); // unref at once - everything is already copied
}
}
return true;
@@ -736,12 +737,6 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
if (rc->send_done_pos == rc->send_out_size)
rc->send_done_pos = 0;
assert(rc->send_done_pos < rc->send_out_size);
while (osd_op_t *op = cl->send_free_ops.front())
{
post_send_free(op);
cl->send_free_ops.pop_front();
}
cl->send_free_ops.pop_front();
if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
{
// Checksums negotiated, enable

View File

@@ -397,7 +397,7 @@ void osd_messenger_t::read_requests()
continue;
}
auto cl = cl_it->second;
if (cl->read_op && cl->read_op_pos >= OSD_PACKET_SIZE && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
if (!rxbounce && cl->read_op && cl->read_op_pos >= OSD_PACKET_SIZE && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
{
get_op_reader_t rdr(this, cl);
if (!op_read_from(cl, rdr))
@@ -677,10 +677,8 @@ bool osd_messenger_t::handle_buffer_with(osd_client_t *cl, uint8_t *curbuf, size
bool osd_messenger_t::handle_hdr(osd_client_t *cl)
{
if (cl->proto_csum_status == MSGR_CSUM_FULL)
{
XXH3_64bits_update(cl->read_csum_state, cl->read_op->req.buf, OSD_PACKET_SIZE);
}
// Note: the header is already checksummed by the reader in op_read_from(),
// which passes RDR_NO_CSUM for it only in MSGR_CSUM_PAYLOAD mode
if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC)
{
auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id);

View File

@@ -307,7 +307,7 @@ class get_op_writer_t: public msgr_op_writer_t
public:
get_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t*, size_t):
msgr(msgr), cl(cl), from(cl->write_op_pos), done(0), enc_size(0), done_enc(0), enc_buf(NULL)
msgr(msgr), cl(cl), from(cl->write_op_pos), done(0), op_enc(0), enc_size(0), done_enc(0), enc_buf(NULL)
{
}
@@ -328,8 +328,8 @@ public:
enc_size = n < GCM_TMP_BUF_SIZE ? GCM_TMP_BUF_SIZE : n;
enc_buf = (uint8_t*)malloc_or_die(enc_size);
done_enc = 0;
assert(!((size_t)enc_buf & 7));
cl->send_free_ops.push_back((osd_op_t*)((size_t)enc_buf | 1));
assert(cl->send_free.size() < UINT64_MAX - cl->next_send_free);
cl->send_free.push_back({ .buf = enc_buf, .flags = MSGR_SENDP_BUF|MSGR_SENDP_REF });
}
}
@@ -525,6 +525,7 @@ restart:
// Do not copy more data into it, just wait for EPOLLOUT
return true;
}
uint64_t send_free_from = cl->next_send_free + cl->send_free.size();
if (cl->hs)
{
// Send handshake message
@@ -532,8 +533,8 @@ restart:
{
uint8_t *out = cl->hs->get_out();
cl->send_list.push_back((iovec){ .iov_base = out, .iov_len = cl->hs->out_size() });
assert(!((size_t)out & 7));
cl->send_free_ops.push_back((osd_op_t*)((size_t)out | 1));
assert(cl->send_free.size() < UINT64_MAX - cl->next_send_free);
cl->send_free.push_back({ .buf = out, .flags = MSGR_SENDP_BUF|MSGR_SENDP_REF });
cl->hs->reset_out();
}
if (!cl->hs->out_size() && cl->hs->done())
@@ -548,6 +549,7 @@ restart:
copy_ops:
copy_ops_to_with<get_op_writer_t>(cl, NULL, 0);
}
uint64_t send_free_to = cl->next_send_free + cl->send_free.size();
if (cl->io_error)
{
stop_client(cl->client_id);
@@ -580,7 +582,10 @@ copy_ops:
cl->write_msg.msg_iovlen = cl->send_list.size();
cl->refs++;
ring_data_t* data = ((ring_data_t*)sqe->user_data);
data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); };
data->callback = [this, cl, send_free_from, send_free_to](ring_data_t *data)
{
handle_send(data->res, data->prev, data->more, cl, send_free_from, send_free_to);
};
bool use_zc = has_sendmsg_zc && min_zerocopy_send_size >= 0;
if (use_zc && min_zerocopy_send_size > 0 &&
cl->send_list_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size)
@@ -614,7 +619,7 @@ copy_ops:
{
result = -errno;
}
if (!handle_send(result, false, false, cl))
if (!handle_send(result, false, false, cl, cl->next_send_free, send_free_to))
{
return false;
}
@@ -668,17 +673,20 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
next_write_op(cl);
}
osd_op_t *op = cl->write_op;
auto prev_pos = cl->write_op_pos;
if (!op_write_to(cl, wr))
{
if (cl->io_error)
return 0;
if (cl->write_op_pos != prev_pos)
{
// part of the operation is already in the current send batch
cl->send_free.push_back({ .buf = op, .flags = MSGR_SENDP_REF });
}
break;
}
if (!cl->write_op && op->op_type == OSD_OP_IN)
{
// this is a reply, free the op after sending it
cl->send_free_ops.push_back(op);
}
assert(cl->send_free.size() < UINT64_MAX - cl->next_send_free);
cl->send_free.push_back({ .buf = op, .flags = (uint32_t)((op->op_type == OSD_OP_IN ? MSGR_SENDP_IN : MSGR_SENDP_OUT) | MSGR_SENDP_REF) });
}
return wr.get_done();
}
@@ -735,7 +743,7 @@ static void eat_send_list(osd_client_t *cl, size_t n)
}
}
bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl, uint64_t free_from, uint64_t free_to)
{
if (!prev)
{
@@ -765,12 +773,8 @@ bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
{
if (prev)
{
// Second notification - only free a batch of postponed ops
int i = 0;
for (; i < cl->zc_free_list.size() && cl->zc_free_list[i]; i++)
post_send_free(cl->zc_free_list[i]);
if (i > 0)
cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
// Second (zero-copy release) notification
cl->unref_send_free(free_from, free_to);
return true;
}
if (cl->send_list_size <= result)
@@ -792,16 +796,10 @@ bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
cl->send_list_size -= result;
return true;
}
for (auto op: cl->send_free_ops)
if (!more)
{
if (more)
cl->zc_free_list.push_back(op);
else
post_send_free(op);
cl->unref_send_free(free_from, free_to);
}
if (more)
cl->zc_free_list.push_back(NULL); // end marker
cl->send_free_ops.clear();
if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
{
// Checksums negotiated, enable

View File

@@ -44,14 +44,37 @@ void osd_op_t::cancel()
}
}
void osd_messenger_t::post_send_free(osd_op_t *op)
// Mark <from>..<to> in <send_free> as unreferenced, free if it's in the beginning
void osd_client_t::unref_send_free(uint64_t from, uint64_t to)
{
if (!op)
return;
else if (!((size_t)op & 7))
delete op;
else
free((void*)((size_t)op & ~(size_t)7));
assert(from >= next_send_free && to <= next_send_free + send_free.size());
bool first = (from == next_send_free);
auto it = send_free.begin() + (from - next_send_free);
for (uint64_t i = from; i < to; i++, it++)
{
it->flags &= ~MSGR_SENDP_REF;
if (first)
{
if (it->flags & MSGR_SENDP_IN)
delete (osd_op_t*)it->buf;
else if (it->flags & MSGR_SENDP_BUF)
free(it->buf);
}
}
if (first)
{
while (it != send_free.end() && !(it->flags & MSGR_SENDP_REF))
{
if (it->flags & MSGR_SENDP_IN)
delete (osd_op_t*)it->buf;
else if (it->flags & MSGR_SENDP_BUF)
free(it->buf);
to++;
it++;
}
next_send_free += (to - from);
send_free.erase(send_free.begin(), it);
}
}
// force_delete means stop the client anyway, even if there are refs to it in the event loop.
@@ -139,12 +162,23 @@ void osd_messenger_t::stop_client(uint64_t client_id, bool force_delete)
}
if (cl->peer_fd >= 0)
{
int r = shutdown(cl->peer_fd, SHUT_RDWR);
if (r != 0 && errno != ENOTCONN)
if (ringloop)
{
fprintf(stderr, "[OSD %ju] failed to shutdown a socket: %s (code %d)\n", osd_num, strerror(errno), errno);
// Flush unflushed send/recv requests, just in case
ringloop->submit();
}
tfd->set_fd_handler(cl->peer_fd, false, NULL);
clients_by_fd.erase(cl->peer_fd);
close(cl->peer_fd);
cl->peer_fd = -1;
}
#ifdef WITH_RDMA
if (cl->rdma_conn)
{
destroy_rdma_conn(cl->rdma_conn);
cl->rdma_conn = NULL;
}
#endif
cl->refs--;
if (cl->refs <= 0 || force_delete)
{
@@ -156,18 +190,6 @@ void osd_messenger_t::destroy_client(osd_client_t *cl)
{
// Find the item again because it can be invalidated at this point
clients.erase(cl->client_id);
if (cl->peer_fd >= 0)
{
tfd->set_fd_handler(cl->peer_fd, false, NULL);
clients_by_fd.erase(cl->peer_fd);
}
#ifdef WITH_RDMA
if (cl->rdma_conn)
{
destroy_rdma_conn(cl->rdma_conn);
cl->rdma_conn = NULL;
}
#endif
delete cl;
}
@@ -175,13 +197,7 @@ osd_client_t::~osd_client_t()
{
free(in_buf);
in_buf = NULL;
if (peer_fd >= 0)
{
// Close the FD only when the client is actually destroyed
// Which only happens when all references are cleared
close(peer_fd);
peer_fd = -1;
}
assert(peer_fd < 0);
// Then cancel all operations
// Operations have to be canceled only after clearing all references to osd_client_t
// because otherwise their buffers may be still present in io_uring asynchronous requests
@@ -191,16 +207,22 @@ osd_client_t::~osd_client_t()
read_op->cancel();
read_op = NULL;
}
// Delete unsent replies - before cancel_ops() because it may delete outbound operations.
if (write_op)
{
if (write_op->op_type == OSD_OP_IN)
delete write_op;
write_op = NULL;
}
for (auto op: write_ops)
{
if (op->op_type == OSD_OP_IN)
delete op;
}
write_ops.clear();
// Cancel outbound ops
cancel_ops();
for (osd_op_t *op: send_free_ops)
{
osd_messenger_t::post_send_free(op);
}
for (osd_op_t *op: zc_free_list)
{
osd_messenger_t::post_send_free(op);
}
unref_send_free(next_send_free, next_send_free + send_free.size());
if (read_csum_state)
{
XXH3_freeState(read_csum_state);

View File

@@ -329,19 +329,21 @@ static void coroutine_fn vitastor_co_resize(void *opaque)
}
#else
BdrvChild *c;
WITH_GRAPH_RDLOCK_GUARD() {
#if QEMU_VERSION_MAJOR >= 9
assert_bdrv_graph_readable();
assert_bdrv_graph_readable();
#endif
QLIST_FOREACH(c, &bs->parents, next_parent) {
QLIST_FOREACH(c, &bs->parents, next_parent) {
#if QEMU_VERSION_MAJOR >= 6
if (c->klass->resize) {
c->klass->resize(c);
}
if (c->klass->resize) {
c->klass->resize(c);
}
#else
if (c->role->resize) {
c->role->resize(c);
}
if (c->role->resize) {
c->role->resize(c);
}
#endif
}
}
#endif
bdrv_dec_in_flight(bs);

View File

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

View File

@@ -82,6 +82,7 @@ static const char* help_text =
" iimg=<image> Copy from Vitastor image <image>\n"
" if=<file> Copy from file <file>\n"
" oimg=<image> Copy to Vitastor image <image>\n"
" opool=<pool> Pool to create output image if it doesn't exist\n"
" of=<file> Copy to file <file>\n"
" bs=1M Set copy block size\n"
" count=N Copy only N input blocks. If N ends in B it counts bytes, not blocks\n"

View File

@@ -983,6 +983,7 @@ std::function<bool(cli_result_t &)> cli_tool_t::start_dd(json11::Json cfg)
dd->parent = this;
dd->iinfo.iimg = cfg["iimg"].string_value();
dd->oinfo.oimg = cfg["oimg"].string_value();
dd->oinfo.out_pool = cfg["opool"].string_value();
dd->iinfo.ifile = cfg["if"].string_value();
dd->oinfo.ofile = cfg["of"].string_value();
dd->blocksize = parse_size(cfg["bs"].string_value());

View File

@@ -25,7 +25,7 @@
//
// Example:
//
// <parent> - <from> - <layer 2> - <to> - <child 1>
// <parent> - <from> - <layer 2> - <to> - <child 1> - <child 5>
// \ \ \- <child 2>
// \ \- <child 3>
// \-<child 4>
@@ -39,7 +39,8 @@
// 3) Process <child 1>:
// - Merge <from>..<child 1> to <layer 2>
// - Set <layer 2> parent to <parent>
// - Rename <layer 2> to <child 1>
// - Swap <layer 2> and <child 1> names and change <child 5> parent to <layer 2>
// - Remove the data of new <layer 2> (old <child 1>)
// 4) Delete other layers of the chain (<from>, <to>)
struct snap_remover_t
{
@@ -70,6 +71,7 @@ struct snap_remover_t
std::vector<std::string> rebased_images, deleted_images;
std::vector<uint64_t> deleted_ids;
std::string inverse_child_name, inverse_parent_name;
inode_config_t inverse_child_new_cfg, inverse_parent_new_cfg;
cli_result_t result;
bool is_done()
@@ -97,6 +99,8 @@ struct snap_remover_t
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);
@@ -155,18 +159,26 @@ resume_3:
resume_4:
while (!wait_result(4))
return;
// Delete "inverse" child data
start_delete_source(inverse_child);
resume_5:
while (!wait_result(5))
return;
// Delete "inverse" child metadata, rename parent over it,
// and also change parent links of the previous "inverse" child
// Delete "inverse" child metadata, swap it with the parent,
// and also change parent links to the previous "inverse" child
rename_inverse_parent();
if (state == 100)
return;
state = 6;
state = 5;
resume_5:
if (parent->waiting > 0)
return;
// Delete "inverse" child data
start_delete_source(inverse_child);
resume_6:
while (!wait_result(6))
return;
// Finally remove the unneeded "inverse" child, renamed to parent
delete_inode_config(inverse_child);
if (state == 100)
return;
state = 7;
resume_7:
if (parent->waiting > 0)
return;
}
@@ -183,18 +195,18 @@ resume_6:
}
// Mark child as deleted
start_mark_deleted(chain_list[current_child]);
resume_9:
while (!wait_result(9))
resume_8:
while (!wait_result(8))
return;
start_delete_source(chain_list[current_child]);
resume_7:
while (!wait_result(7))
resume_9:
while (!wait_result(9))
return;
delete_inode_config(chain_list[current_child]);
if (state == 100)
return;
state = 8;
resume_8:
state = 10;
resume_10:
if (parent->waiting > 0)
return;
}
@@ -461,11 +473,13 @@ resume_100:
parent->cli->st_cli->etcd_prefix+"/index/image/"+inverse_parent_name
);
// Fill new configuration
inode_config_t new_cfg = *child_cfg;
new_cfg.deleted = false;
new_cfg.num = target_cfg->num;
new_cfg.enc_key = target_cfg->enc_key;
new_cfg.parent_id = new_parent;
inverse_parent_new_cfg = *child_cfg;
inverse_parent_new_cfg.deleted = false;
inverse_parent_new_cfg.num = target_cfg->num;
inverse_parent_new_cfg.enc_key = target_cfg->enc_key;
inverse_parent_new_cfg.parent_id = new_parent;
inverse_child_new_cfg = *child_cfg;
inverse_child_new_cfg.name = inverse_parent_name;
json11::Json::array cmp = json11::Json::array {
json11::Json::object {
{ "target", "MOD" },
@@ -481,20 +495,27 @@ resume_100:
},
};
json11::Json::array txn = json11::Json::array {
// child -> rename to parent
json11::Json::object {
{ "request_delete_range", json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", child_cfg_key },
} },
},
json11::Json::object {
{ "request_delete_range", json11::Json::object {
{ "key", target_idx_key },
{ "value", base64_encode(json11::Json(parent->cli->st_cli->serialize_inode_cfg(&inverse_child_new_cfg)).dump()) },
} },
},
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", target_idx_key },
{ "value", base64_encode(json11::Json({
{ "id", INODE_NO_POOL(inverse_child) },
{ "pool_id", (uint64_t)INODE_POOL(inverse_child) },
}).dump()) },
} },
},
// parent -> rename to child
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", target_cfg_key },
{ "value", base64_encode(json11::Json(parent->cli->st_cli->serialize_inode_cfg(&new_cfg)).dump()) },
{ "value", base64_encode(json11::Json(parent->cli->st_cli->serialize_inode_cfg(&inverse_parent_new_cfg)).dump()) },
} },
},
json11::Json::object {
@@ -556,8 +577,14 @@ resume_100:
state = 100;
return;
}
inverse_child_new_cfg.mod_revision = res["header"]["revision"].uint64_value();
inverse_parent_new_cfg.mod_revision = res["header"]["revision"].uint64_value();
parent->cli->st_cli->inode_by_name.erase(inverse_child_name);
parent->cli->st_cli->inode_by_name.erase(inverse_parent_name);
parent->cli->st_cli->insert_inode_config(inverse_child_new_cfg);
parent->cli->st_cli->insert_inode_config(inverse_parent_new_cfg);
if (parent->progress)
printf("Layer %s renamed to %s\n", inverse_parent_name.c_str(), inverse_child_name.c_str());
printf("Layers %s and %s swapped\n", inverse_parent_name.c_str(), inverse_child_name.c_str());
parent->ringloop->wakeup();
});
}

View File

@@ -1,7 +1,17 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#if defined(__GNUC__) && !defined(__clang__)
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
#endif
#include <regex>
#if defined(__GNUC__) && !defined(__clang__)
# pragma GCC diagnostic pop
#endif
#include "disk_tool.h"
#include "str_util.h"

View File

@@ -26,6 +26,10 @@ int main(int argc, char *argv[])
{
pg_t pg = {
.state = PG_PEERING,
.scheme = POOL_SCHEME_XOR,
.pg_cursize = 3,
.pg_size = 3,
.pg_minsize = 2,
.pg_num = 1,
.target_set = { 1, 2, 3 },
.cur_set = { 1, 2, 3 },

View File

@@ -309,29 +309,6 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
{
osd_op_t *cur_op = (osd_op_t*)subop->op_type;
blockstore_op_t *bs_op = subop->bs_op;
int64_t expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
|| bs_op->opcode == BS_OP_WRITE_STABLE ? bs_op->len : 0;
if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ && bs_op->retval != -EAGAIN &&
(bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE || bs_op->retval != -ENOSPC))
{
// die on any error except ENOSPC or EAGAIN 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 = %jd (expected %jd)\n",
osd_op_names[bs_op_to_osd_op[bs_op->opcode]],
bs_op->oid.inode, bs_op->oid.stripe, bs_op->version, bs_op->retval, expected
);
}
else
{
printf(
"%s subop failed locally: retval = %jd (expected %jd)\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->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB;
add_bs_subop_stats(subop, recovery_related);
subop->req.hdr.opcode = bs_op_to_osd_op[bs_op->opcode];

View File

@@ -1788,6 +1788,134 @@ static void test_compact_intent_write_in_partial_csum_block()
free(buf);
}
// Placing an intent write over big_write is always OK except in one case:
// if the big_write is a result of unfinished compaction and has a preceding small_write entry.
// Because in this case the compaction big_write may disappear after a reboot and small_write
// can then be replayed over our intent_write. However, if the big_write only has intent_writes
// before it - overwriting it again is OK.
static void test_no_intent_over_unwritten_compaction()
{
printf("\n-- test_no_intent_over_unwritten_compaction\n");
bs_test_t test;
test.default_cfg();
test.init();
object_id oid = { .inode = 1, .stripe = 0 };
blockstore_op_t op;
op.opcode = BS_OP_WRITE_STABLE;
op.oid = oid;
op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
// A full block, then a 4K write over it - which is done in place because the big_write
// under it is already on the disk
printf("writing 0+128K v1\n");
op.version = 1;
op.offset = 0;
op.len = 128*1024;
memset(op.buf, 0xaa, op.len);
test.exec_op(&op);
assert(op.retval == op.len);
printf("writing 4K+4K v2\n");
op.version = 2;
op.offset = 4096;
op.len = 4096;
memset(op.buf, 0xbb, op.len);
test.exec_op(&op);
assert(op.retval == op.len);
auto obj = test.bs->heap->read_entry(oid);
assert(obj && obj->entry_type == (BS_HEAP_BIG_INTENT|BS_HEAP_STABLE));
// 8K is over atomic_write_size, so this one can't be done in place and goes to the buffer
printf("writing 16K+8K v3\n");
op.version = 3;
op.offset = 16384;
op.len = 8192;
memset(op.buf, 0xcc, op.len);
test.exec_op(&op);
assert(op.retval == op.len);
obj = test.bs->heap->read_entry(oid);
assert(obj && obj->entry_type == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE));
// Compact the object, but hold the metadata write of the compacted entry. Skip the
// superblock - it's written by the same flusher and holding it would stall everything
io_uring_sqe held_sqe;
ring_data_t *held_data = NULL;
test.sqe_handler = [&](io_uring_sqe *sqe)
{
if (!held_data && sqe->fd == MOCK_DATA_FD && sqe->opcode == IORING_OP_WRITEV &&
sqe->off >= test.bs->dsk.meta_offset + test.bs->dsk.meta_block_size &&
sqe->off < parse_size(test.config["journal_offset"]))
{
held_sqe = *sqe;
held_data = (ring_data_t*)sqe->user_data;
return true;
}
return false;
};
printf("compacting\n");
test.bs->flusher->request_trim();
while (!held_data)
test.ringloop->loop();
obj = test.bs->heap->read_entry(oid);
assert(obj && obj->entry_type == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE));
// Another 4K write. It looks like a perfect candidate for an in-place write - a big_write
// with an old LSN right under it - but that entry isn't on the disk yet, so it has to be
// buffered instead
printf("writing 40K+4K v4\n");
op.version = 4;
op.offset = 40960;
op.len = 4096;
memset(op.buf, 0xdd, op.len);
test.exec_op(&op);
assert(op.retval == op.len);
obj = test.bs->heap->read_entry(oid);
assert(obj && obj->entry_type == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE));
// Let the compaction finish
printf("releasing the metadata write\n");
bool ok = test.data_disk->submit(&held_sqe);
assert(ok);
test.ringloop->mark_completed(held_data);
test.sqe_handler = nullptr;
while (test.bs->flusher->is_active())
test.ringloop->loop();
test.bs->flusher->release_trim();
// Everything must read back, both now and after a restart
blockstore_op_t op2;
op2.opcode = BS_OP_READ;
op2.oid = oid;
op2.version = UINT64_MAX;
op2.offset = 0;
op2.len = 128*1024;
op2.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
for (int i = 0; i < 2; i++)
{
if (i)
{
printf("restarting\n");
test.destroy_bs();
test.init();
}
printf("reading\n");
test.exec_op(&op2);
assert(op2.retval == op2.len);
assert(memcheck(op2.buf, 0xaa, 4096));
assert(memcheck(op2.buf+4096, 0xbb, 4096));
assert(memcheck(op2.buf+8192, 0xaa, 8192));
assert(memcheck(op2.buf+16384, 0xcc, 8192));
assert(memcheck(op2.buf+24576, 0xaa, 16384));
assert(memcheck(op2.buf+40960, 0xdd, 4096));
assert(memcheck(op2.buf+45056, 0xaa, 128*1024-45056));
}
free(op.buf);
free(op2.buf);
}
int main(int narg, char *args[])
{
test_simple();
@@ -1815,5 +1943,6 @@ int main(int narg, char *args[])
test_recheck_under_commit();
test_compact_over_rolled_back_big_write();
test_compact_intent_write_in_partial_csum_block();
test_no_intent_over_unwritten_compaction();
return 0;
}

View File

@@ -174,8 +174,17 @@ struct bs_test_t
done = true;
};
bs->enqueue_op(op);
int iters = 0;
while (!done)
{
ringloop->loop();
iters++;
if (iters >= 100000)
{
fprintf(stderr, "Operation didn't complete in %d iterations - deadlock?\n", iters);
abort();
}
}
op->callback = nullptr;
}
};
@@ -2648,6 +2657,153 @@ static void test_no_double_submit_of_journal_sector()
free(buf);
}
static void test_rollback_over_stable_write()
{
printf("\n-- test_rollback_over_stable_write\n");
bs_test_t test;
test.default_cfg();
test.config["disable_data_fsync"] = "0";
test.config["immediate_commit"] = "none";
test.init();
printf("blockstore initialized\n");
object_id oid = { .inode = 1, .stripe = 0 };
uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
blockstore_op_t op;
op.oid = oid;
op.buf = buf;
printf("write v1\n");
op.opcode = BS_OP_WRITE;
op.version = 1;
op.offset = 0;
op.len = 4096;
memset(buf, 0xAA, 4096);
test.exec_op(&op);
assert(op.retval == (int)op.len);
printf("write v2\n");
op.opcode = BS_OP_WRITE;
op.version = 2;
op.offset = 0;
op.len = 4096;
memset(buf, 0xEF, 4096);
test.exec_op(&op);
assert(op.retval == (int)op.len);
// commit v2
printf("sync+stabilize v2\n");
op.opcode = BS_OP_SYNC;
test.exec_op(&op);
assert(op.retval == 0);
blockstore_op_t st;
st.opcode = BS_OP_STABLE;
st.len = 1;
st.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, sizeof(obj_ver_id));
*((obj_ver_id*)st.buf) = (obj_ver_id){ .oid = oid, .version = 2 };
test.exec_op(&st);
assert(st.retval == 0);
printf("try to roll back to v1 - should complete with EBUSY\n");
st.opcode = BS_OP_ROLLBACK;
*((obj_ver_id*)st.buf) = (obj_ver_id){ .oid = oid, .version = 1 };
test.exec_op(&st);
assert(st.retval == -EBUSY);
printf("...got EBUSY\n");
free(st.buf);
free(buf);
}
// A deletion takes the preceding unsynced writes of the object over: it drops them from the
// unsynced lists and erases them from the journal, because from that moment the only thing on
// the disk accounting for them is the deletion itself. A sync submitted while the deletion is
// still in flight must therefore wait for it - otherwise it finds nothing left to sync, reports
// success, and a power outage then takes away both the deletion and the writes it swallowed,
// leaving the object at a version older than the one the sync acknowledged
static void test_sync_waits_for_delete_taking_over_unsynced()
{
printf("\n-- test_sync_waits_for_delete_taking_over_unsynced\n");
bs_test_t test;
test.default_cfg();
// Writes are only durable after an explicit sync here
test.config["disable_data_fsync"] = "0";
test.config["immediate_commit"] = "none";
test.init();
printf("blockstore initialized\n");
object_id oid = { .inode = 1, .stripe = 0 };
uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
printf("write v1 and make it durable\n");
blockstore_op_t op;
op.opcode = BS_OP_WRITE_STABLE;
op.oid = oid;
op.version = 1;
op.offset = 0;
op.len = 4096;
op.buf = buf;
memset(buf, 0xAA, 4096);
test.exec_op(&op);
assert(op.retval == (int)op.len);
blockstore_op_t sync0;
sync0.opcode = BS_OP_SYNC;
sync0.buf = NULL;
test.exec_op(&sync0);
assert(sync0.retval == 0);
printf("write v2 (acknowledged, but unsynced)\n");
memset(buf, 0xBB, 4096);
op.version = 2;
test.exec_op(&op);
assert(op.retval == (int)op.len);
printf("submitting a deletion and a sync in one batch\n");
blockstore_op_t del;
bool del_done = false;
del.opcode = BS_OP_DELETE;
del.oid = oid;
del.version = 3;
del.buf = NULL;
del.callback = [&](blockstore_op_t *op) { del_done = true; };
blockstore_op_t sync;
bool sync_done = false;
sync.opcode = BS_OP_SYNC;
sync.buf = NULL;
sync.callback = [&](blockstore_op_t *op) { sync_done = true; };
test.bs->enqueue_op(&del);
test.bs->enqueue_op(&sync);
while (!sync_done)
test.ringloop->loop();
assert(sync.retval == 0);
// The deletion swallowed v2, so the sync could only report success after it
assert(del_done);
assert(del.retval == 0);
printf("now the power goes away - the volatile cache is lost\n");
test.destroy_bs();
test.ringloop->reset();
test.data_disk->discard_buffers(true, 0);
test.init();
printf("the object must be gone, not back at v1\n");
blockstore_op_t rd;
rd.opcode = BS_OP_READ;
rd.oid = oid;
rd.version = UINT64_MAX;
rd.offset = 0;
rd.len = 4096;
rd.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, rd.len);
test.exec_op(&rd);
assert(rd.retval == -ENOENT);
free(rd.buf);
free(buf);
}
int main(int narg, char *args[])
{
test_preserve_corruption();
@@ -2678,5 +2834,7 @@ int main(int narg, char *args[])
test_ack_waits_for_earlier_journal_data();
test_no_double_submit_of_journal_sector();
test_recovered_state_is_made_durable();
test_rollback_over_stable_write();
test_sync_waits_for_delete_taking_over_unsynced();
return 0;
}

View File

@@ -351,6 +351,8 @@ resume_2:
int main(int narg, char *args[])
{
setvbuf(stdout, NULL, _IONBF, 0);
setvbuf(stderr, NULL, _IONBF, 0);
json11::Json::object cfgo;
for (int i = 1; i < narg; i++)
{

View File

@@ -1082,6 +1082,44 @@ void test_iterate_compaction()
assert(small_writes == 0);
}
printf("-----\n");
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// Case: BIG(v1 l1) SMALL(v2 l2) COMMIT(v1 l3) SMALL(v3 l4) COMMIT(v2 l5)
// -> compact by adding BIG_STABLE(v2 l3) (!) and skip l4 and l5
uint32_t mblock = 0;
_test_big_write(heap, dsk, 1, 0, 1, 0, false, 0, 4096, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 4*1024, 4096, 4*1024, false, buffer_area.data(), false);
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
auto obj = heap.read_entry(oid);
res = heap.add_commit(obj, 1, &mblock);
assert(res == 0);
heap.start_block_write(mblock);
heap.complete_block_write(mblock);
_test_small_write(heap, dsk, 1, 0, 3, 12*1024, 4096, 12*1024, false, buffer_area.data(), false);
obj = heap.read_entry(oid);
res = heap.add_commit(obj, 2, &mblock);
assert(res == 0);
heap.start_block_write(mblock);
heap.complete_block_write(mblock);
assert(heap.get_fsynced_lsn() == 5);
obj = heap.read_entry(oid);
assert(count_writes(heap, obj) == 5);
int small_writes = 0;
auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
{
small_writes++;
});
assert(small_writes == 1);
assert(compact_info.compact_lsn == 3);
assert(compact_info.compact_version == 2);
assert(compact_info.clean_wr->lsn == 1);
assert(!compact_info.do_delete);
}
printf("OK test_iterate_compaction\n");
}
@@ -1224,6 +1262,11 @@ void test_recheck(bool async, bool csum)
memcpy(buf, buffer_area.data(), len);
else if (offset == 0x20000+8*1024) // invalid
memset(buf, 0xcc, len);
// Object 2's older intent write: dropping the newest one exposes it, and
// the newest is exactly what an in-place write may have destroyed, so it
// gets rechecked too. Here it's intact
else if (offset == 0x20000+4*1024)
memcpy(buf, buffer_area.data(), len);
else if (offset == 0xA0000+16*1024) // valid
memcpy(buf, buffer_area.data(), len);
else if (offset == 0xC0000+16*1024) // invalid
@@ -1247,7 +1290,7 @@ void test_recheck(bool async, bool csum)
}
}, 1);
assert(done);
assert(calls == (async ? 13 : 7));
assert(calls == (async ? 14 : 8));
heap.finish_recheck();
@@ -1588,6 +1631,216 @@ void test_recheck_skips_compacted_parts(bool async)
printf("OK test_recheck_skips_compacted_parts\n");
}
// Dropping an unfinished entry uncovers the entries below it, and their data may be just as
// unfinished - a buffered small write whose data never landed, or another redirect intent.
// They become the newest entries of the object, so they have to be rechecked in turn.
void test_recheck_under_dropped_intent(bool async)
{
printf("test_recheck_under_dropped_intent %s\n", async ? "async" : "sync");
blockstore_disk_t dsk;
_test_init(dsk, true);
// Keep the metadata fsync enabled so that adding the redirect intent doesn't immediately
// mark the entries under it as garbage: the block persisted below is the one an outage
// leaves behind, written when the intent was added and never rewritten since
dsk.disable_meta_fsync = false;
std::vector<uint8_t> buffer_area(dsk.journal_device_size);
std::vector<uint8_t> tmp;
memset(buffer_area.data(), 0xab, 32*1024);
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// Both objects are a big_write with a buffered small write and a redirect intent
// on top of it, and in both the intent's data never reached the disk
_test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 8192, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 4096, 8192, 0, true, buffer_area.data());
_test_redirect_intent(heap, dsk, 1, 0, 3, 0x20000, true, 16384, 8192, buffer_area.data());
_test_big_write(heap, dsk, 2, 0, 1, 0x40000, true, 0, 8192, buffer_area.data());
_test_small_write(heap, dsk, 2, 0, 2, 4096, 8192, 8192, true, buffer_area.data());
_test_redirect_intent(heap, dsk, 2, 0, 3, 0x60000, true, 16384, 8192, buffer_area.data());
tmp.resize(dsk.meta_block_size);
heap.get_meta_block(0, tmp.data());
}
// Object 1 also lost the buffered data of the small write which the intent was hiding
memset(buffer_area.data(), 0xcc, 8192);
{
blockstore_heap_t heap(&dsk, async ? NULL : buffer_area.data(), 10);
uint64_t entries_loaded;
heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
heap.finish_load();
bool done = heap.recheck_small_writes([&](bool is_data, uint64_t offset, uint64_t len, std::function<void(uint8_t *buf)> cb)
{
if (!len)
{
return;
}
uint8_t *buf = (uint8_t*)malloc_or_die(len);
// Neither intent's data landed
if (is_data)
memset(buf, 0xcc, len);
else
memcpy(buf, buffer_area.data()+offset, len);
assert(cb);
cb(buf);
free(buf);
}, 1);
assert(done);
heap.finish_recheck();
// Object 1 is left with the big_write alone - the small write uncovered by dropping
// the intent is unfinished as well
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
heap_entry_t *obj = heap.read_entry(oid);
assert(obj);
assert(count_writes(heap, obj) == 1);
assert(obj->entry_type == BS_HEAP_BIG_WRITE|BS_HEAP_STABLE);
assert(obj->version == 1);
// Object 2 keeps its small write - only the intent above it was unfinished
oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 };
obj = heap.read_entry(oid);
assert(obj);
assert(count_writes(heap, obj) == 2);
assert(obj->entry_type == BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE);
assert(obj->version == 2);
assert(check_used_space(heap, dsk, 0));
}
printf("OK test_recheck_under_dropped_intent\n");
}
// A part of an entry is only shadowed as long as the entry shadowing it stays. The recheck may
// find that newer entry unfinished and drop it, and then the bytes under it become visible
// again - so they have to be verified after all, and the entry dropped as well if they are bad.
// <reverse> completes the reads in the opposite order to make sure the recheck doesn't depend
// on the order in which the disk answers.
void test_recheck_shadow_of_dropped_entry(bool reverse)
{
printf("test_recheck_shadow_of_dropped_entry %s\n", reverse ? "reverse" : "in order");
blockstore_disk_t dsk;
_test_init(dsk, true);
// Skipping is only exact while a checksum block isn't larger than the write granularity
assert(dsk.csum_block_size == dsk.bitmap_granularity);
std::vector<uint8_t> buffer_area(dsk.journal_device_size);
std::vector<uint8_t> tmp;
memset(buffer_area.data(), 0xab, 32*1024);
{
blockstore_heap_t heap(&dsk, buffer_area.data());
heap.finish_recheck();
// All three objects are a big_intent covering 0..16K with a small write shadowing 4K..8K
_test_redirect_intent(heap, dsk, 1, 0, 1, 0, true, 0, 16384, buffer_area.data());
_test_small_write(heap, dsk, 1, 0, 2, 4096, 4096, 0, true, buffer_area.data());
_test_redirect_intent(heap, dsk, 2, 0, 1, 0x20000, true, 0, 16384, buffer_area.data());
_test_small_write(heap, dsk, 2, 0, 2, 4096, 4096, 4096, true, buffer_area.data());
_test_redirect_intent(heap, dsk, 3, 0, 1, 0x40000, true, 0, 16384, buffer_area.data());
_test_small_write(heap, dsk, 3, 0, 2, 4096, 4096, 8192, true, buffer_area.data());
tmp.resize(dsk.meta_block_size);
heap.get_meta_block(0, tmp.data());
}
{
blockstore_heap_t heap(&dsk, NULL, 10);
uint64_t entries_loaded;
heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
heap.finish_load();
struct pending_read_t
{
bool is_data;
uint64_t offset, len;
std::function<void(uint8_t*)> cb;
};
std::vector<pending_read_t> pending;
bool all_done = false;
auto reader = [&](bool is_data, uint64_t offset, uint64_t len, std::function<void(uint8_t *buf)> cb)
{
if (!len)
{
all_done = true;
return;
}
pending.push_back((pending_read_t){ .is_data = is_data, .offset = offset, .len = len, .cb = cb });
};
auto complete = [&](const pending_read_t & rd)
{
uint8_t *buf = (uint8_t*)malloc_or_die(rd.len);
memset(buf, 0xab, rd.len);
if (rd.is_data)
{
// Object 1: the shadowed 4K..8K of the base block is wrong, and so is the
// small write shadowing it - so the base entry isn't verified by anything
if (rd.offset <= 4096 && rd.offset+rd.len >= 8192)
memset(buf + 4096 - rd.offset, 0xcc, 4096);
// Object 3: only the shadowed part is wrong, and the small write over it is
// good - which is what an in-place compaction looks like
if (rd.offset <= 0x40000+4096 && rd.offset+rd.len >= 0x40000+8192)
memset(buf + 0x40000 + 4096 - rd.offset, 0xcc, 4096);
}
else
{
// Buffered data of the small writes of objects 1 and 2 never landed
if (rd.offset < 8192)
memset(buf, 0xcc, rd.len);
}
assert(rd.cb);
rd.cb(buf);
free(buf);
};
heap.recheck_small_writes(reader, 4);
while (!all_done)
{
assert(pending.size() > 0);
auto batch = pending;
pending.clear();
for (size_t i = 0; i < batch.size(); i++)
complete(batch[reverse ? batch.size()-1-i : i]);
}
heap.finish_recheck();
// Object 1 is rolled back completely: the small write is gone and the part of the
// base entry it used to shadow turned out to be bad
object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
heap_entry_t *obj = heap.read_entry(oid);
assert(!obj);
// Object 2 only loses the small write - the base entry under it is intact
oid = { .inode = INODE_WITH_POOL(1, 2), .stripe = 0 };
obj = heap.read_entry(oid);
assert(obj);
assert(count_writes(heap, obj) == 1);
assert(obj->version == 1);
// Object 3 survives whole - the mismatch is in a part which stays shadowed
oid = { .inode = INODE_WITH_POOL(1, 3), .stripe = 0 };
obj = heap.read_entry(oid);
assert(obj);
assert(count_writes(heap, obj) == 2);
assert(obj->version == 2);
assert(check_used_space(heap, dsk, 0));
}
printf("OK test_recheck_shadow_of_dropped_entry\n");
}
void test_corruption()
{
blockstore_disk_t dsk;
@@ -3150,6 +3403,10 @@ int main(int narg, char *args[])
test_recheck_intent_under_small_writes(true, false);
test_recheck_skips_compacted_parts(false);
test_recheck_skips_compacted_parts(true);
test_recheck_shadow_of_dropped_entry(false);
test_recheck_shadow_of_dropped_entry(true);
test_recheck_under_dropped_intent(false);
test_recheck_under_dropped_intent(true);
test_entry_placement_contract();
test_corruption();
test_full_overwrite(true);

View File

@@ -23,14 +23,9 @@ ring_loop_t::ring_loop_t(int qd, bool multithreaded, bool sqe128)
{
throw std::runtime_error(std::string("io_uring_queue_init: ") + strerror(-ret));
}
free_ring_data_ptr = *ring.sq.kring_entries;
ring_datas = (struct ring_data_t*)calloc(free_ring_data_ptr, sizeof(ring_data_t));
free_ring_data = (int*)malloc(sizeof(int) * free_ring_data_ptr);
if (!ring_datas || !free_ring_data)
{
throw std::bad_alloc();
}
for (int i = 0; i < free_ring_data_ptr; i++)
ring_datas.resize(*ring.sq.kring_entries);
free_ring_data.resize(*ring.sq.kring_entries);
for (int i = 0; i < free_ring_data.size(); i++)
{
free_ring_data[i] = i;
}
@@ -46,8 +41,6 @@ ring_loop_t::ring_loop_t(int qd, bool multithreaded, bool sqe128)
ring_loop_t::~ring_loop_t()
{
free(free_ring_data);
free(ring_datas);
io_uring_queue_exit(&ring);
if (ring_eventfd)
{
@@ -82,7 +75,7 @@ io_uring_sqe* ring_loop_t::get_sqe()
{
if (mt)
mu.lock();
if (free_ring_data_ptr == 0)
if (!free_ring_data.size())
{
if (mt)
mu.unlock();
@@ -91,7 +84,8 @@ io_uring_sqe* ring_loop_t::get_sqe()
struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);
assert(sqe);
*sqe = { 0 };
io_uring_sqe_set_data(sqe, ring_datas + free_ring_data[--free_ring_data_ptr]);
io_uring_sqe_set_data(sqe, ring_datas.data() + free_ring_data.back());
free_ring_data.pop_back();
if (mt)
mu.unlock();
return sqe;
@@ -144,7 +138,7 @@ void ring_loop_t::loop()
dl.prev = d->prev;
dl.callback.swap(d->callback);
d->prev = d->more = false;
free_ring_data[free_ring_data_ptr++] = d - ring_datas;
free_ring_data.push_back(d - ring_datas.data());
if (mt)
mu.unlock();
dl.callback(&dl);
@@ -152,7 +146,7 @@ void ring_loop_t::loop()
else
{
fprintf(stderr, "Warning: empty callback in SQE\n");
free_ring_data[free_ring_data_ptr++] = d - ring_datas;
free_ring_data.push_back(d - ring_datas.data());
if (mt)
mu.unlock();
}
@@ -187,7 +181,7 @@ void ring_loop_t::restore(unsigned sqe_tail)
unsigned inc = (1 << io_uring_sqe_shift(&ring));
for (unsigned i = sqe_tail; i < ring.sq.sqe_tail; i += inc)
{
free_ring_data[free_ring_data_ptr++] = ((ring_data_t*)ring.sq.sqes[i & *ring.sq.kring_mask].user_data) - ring_datas;
free_ring_data.push_back(((ring_data_t*)ring.sq.sqes[i & *ring.sq.kring_mask].user_data) - ring_datas.data());
}
ring.sq.sqe_tail = sqe_tail;
}
@@ -198,10 +192,10 @@ unsigned ring_loop_t::space_left()
unsigned int head = io_uring_smp_load_acquire(sq->khead);
unsigned int next = sq->sqe_tail + 1;
int left = (*sq->kring_entries - (next - head)) >> io_uring_sqe_shift(&ring);
if (left > free_ring_data_ptr)
if (left > free_ring_data.size())
{
// return min(sqes left, ring_datas left)
return free_ring_data_ptr;
return free_ring_data.size();
}
return left;
}

View File

@@ -58,11 +58,10 @@ class __attribute__((visibility("default"))) ring_loop_t: public ring_loop_i
{
std::vector<std::function<void()>> immediate_queue, immediate_queue2;
std::vector<ring_consumer_t*> consumers;
struct ring_data_t *ring_datas;
std::vector<ring_data_t> ring_datas;
std::vector<int> free_ring_data;
std::mutex mu;
bool mt;
int *free_ring_data;
unsigned free_ring_data_ptr;
bool in_loop;
bool loop_again;
struct io_uring ring;