Compare commits

...
21 Commits
Author SHA1 Message Date
Antonio S. Musumeci 76d3990776 Add fuse-read-spin-us: poll /dev/fuse briefly before blocking
After handling a request, a sync read thread polls its /dev/fuse fd for
up to fuse-read-spin-us microseconds before blocking in read(). Default
0 (disabled). Measured at 50us (tmpfs branch, direct_io): sequential
128K/1M reads +27-32%, 128K writes +14%, metadata walks +24-32%; no
change on 8 parallel streams or 1M writes. Costs roughly one core on a
busy-but-quiet mount, nothing when idle.
2026-10-14 19:55:12 -04:00
Antonio S. Musumeci 54166afe08 splice 2026-10-10 14:32:40 -04:00
Antonio SJ Musumeci 515b0549ef docs/performance: explain mergerfs IO performance vs RAID
Describe why mergerfs does not aggregate branch performance like
RAID/ZFS (whose aggregate throughput is also capped by shared
interconnects) and instead provides independent per-branch
performance: best case A+B+C across concurrent streams, worst case
single-stream limited to one branch's speed.
2026-09-05 16:00:43 -04:00
natalefalco 8ea47df13e Include Docker Compose setup for mergerfs
Added Docker Compose configuration for running mergerfs.
2026-07-31 08:13:59 -05:00
shyney7 42ab831d97 docs/setup: link install :z example to SELinux FAQ 2026-05-17 10:14:21 -05:00
shyney7 89b4046d11 docs/faq: add SELinux relabeling entry 2026-05-17 10:14:21 -05:00
Antonio SJ Musumeci 7ec6f8e2f2 Manage cloned /dev/fuse fds in sessions
- Store cloned read-worker fds on fuse_session and close them there.
- Set up clone fallback through session helpers used by the loop.
- Carry per-request fds through receive, process, and reply paths.

Tests show significant performance increases in highly concurrent
workloads on a high core count system. Upto 15% better on reads and
27% on writes.
2026-05-14 20:13:07 -05:00
Antonio SJ Musumeci 5f7fa17ad8 Inline ioprio disabled fast path
- Expose the ioprio enabled flag as an atomic for relaxed inline checks
- Inline SetFrom construction so disabled read/write paths avoid a function call
- Move priority application into a slow helper used only when ioprio is enabled
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 30d59d4458 Use shared_mutex for fuse log config
- Guard fuse_cfg_t log file and filepath shared pointers with a shared mutex
- Replace atomic shared pointer loads and stores with shared and unique locks
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 02dc36080c Use shared_mutex for ReadDir state
- Replace atomic shared pointer loads/stores with shared/unique locking.
- Store the configured readdir mode directly as std::string.
- Copy the active ReadDir implementation under lock before invoking it.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 8734847968 Guard branch snapshots with shared_mutex
Important changes:
- Replace atomic shared_ptr operations with shared_mutex-protected snapshots.
- Retry copy-on-write branch updates when another writer publishes first.
- Keep long-running parsing and read-only filesystem checks outside branch locks.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 608cd00764 Use atomic open file reference counts
- Use relaxed increment under the map bucket lock when duplicating file handles.
- Use acq_rel decrement when releasing and checking for final references.
- Load moved ref_count with relaxed ordering during map insertion.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci b13727abde Document readdir merge mutex scope
- Explain that the mutex is per readdir call.
- Document that only dirent deduplication and list append are serialized.
- Clarify that per-branch opendir and getdents64 work runs outside the critical section.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 231aac15cd Use shared write locks until ENOSPC relocation
- Replace the per-file Mutex with std::shared_mutex for shared write access.
- Run direct and cached writes under a shared lock while the fd is stable.
- Retry failed writes under a unique lock before move-on-ENOSPC relocation.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 8b17d5f35e Synchronize node updates with path release
- Add helpers that release path locks and update node state under f.lock.
- Update getattr and setattr stat-cache invalidation while preserving ESTALE handling.
- Rework create, open, tmpfile, and set_path_info to mutate lookup/open state under the shared lock.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 8608b2be46 Clarify releasedir readdir synchronization barrier
- Document the lock/unlock pair as an acquire/release barrier.
- Explain that releasedir waits for in-flight readdir calls before destroy/free.
- Warn against removing the synchronization despite it not being a critical section.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 1e0bd54989 Move statvfs cache to use shared mutex
Its read heavy and micro benchmarks show no regression on the low end
and 30%+ throughput improvements on the highend.
2026-05-14 00:22:06 -04:00
Antonio SJ Musumeci 56285fc6e5 Pass annobin override to RPM toolset builds
Important changes:
- Add Makefile RPMBUILD and RPMBUILD_FLAGS overrides for the rpm target.
- Detect /opt/rh GCC toolsets before RPM packaging.
- Pass --undefine _annotated_build to avoid Rocky 8 annobin plugin failures.
2026-05-13 17:01:41 -04:00
Antonio SJ Musumeci dfe053a0ae Require C++20-capable build toolchains
Important changes:
- Switch mergerfs and vendored libfuse Makefiles to -std=c++20.
- Select the latest available GCC/G++ or toolset in container builds.
- Verify compiler C++20 support during package installation.
- Fix dynamic SysLog formatting for fmt C++20 checks.
2026-05-13 13:55:43 -05:00
Antonio SJ Musumeci 12ac368e8b Preserve CPPFLAGS in release builds
- Append -DNDEBUG to existing CPPFLAGS in the release target.
- Keep caller-provided build flags for Debian packaging.
2026-05-09 19:28:56 -05:00
Antonio SJ Musumeci c8f32f9f25 update build system and dependency configuration 2026-05-09 18:55:23 -05:00
50 changed files with 3060 additions and 385 deletions
+31 -22
View File
@@ -29,6 +29,8 @@ MKDIR ?= mkdir
MKTEMP ?= mktemp
MV ?= mv
RM ?= rm
RPMBUILD ?= rpmbuild
RPMBUILD_FLAGS ?=
SED ?= sed
STRIP ?= strip
TAR ?= tar
@@ -37,7 +39,7 @@ TOUCH ?= touch
BUILDDIR := build
GITHUB_REPO ?= "https://github.com/trapexit/mergerfs"
DEFAULT_RELEASE := debian:13.amd64
DEFAULT_TARGET := debian:13.amd64
ifndef GIT_REPO
ifneq ($(shell $(GIT) --version 2> /dev/null),)
@@ -48,11 +50,7 @@ endif
endif
USE_XATTR ?= 1
UGID_USE_RWLOCK ?= 0
ifdef NDEBUG
OPT_FLAGS := -O2 -DNDEBUG
else
ifdef SANITIZE
ifeq ($(SANITIZE),1)
override SANITIZE := -fsanitize=address,undefined,leak
@@ -60,14 +58,22 @@ else ifeq ($(filter -fsanitize=%,$(SANITIZE)),)
override SANITIZE := -fsanitize=$(SANITIZE)
endif
endif
OPT_FLAGS := -O0 \
-g \
-fno-omit-frame-pointer \
$(SANITIZE) \
-fstack-protector-strong \
-Wextra \
-Wno-unused-parameter \
-DDEBUG
COMMON_OPT_FLAGS += \
-fstack-protector-strong \
-Wextra \
-Wno-unused-parameter
DEBUG_OPT_FLAGS += \
-O0 \
-g3 \
-fno-omit-frame-pointer \
$(SANITIZE)
ifdef RELEASE
OPT_FLAGS := -O3 $(COMMON_OPT_FLAGS)
else
OPT_FLAGS ?= $(DEBUG_OPT_FLAGS) $(COMMON_OPT_FLAGS)
endif
ifdef STATIC
@@ -117,7 +123,7 @@ CXXFLAGS ?= \
-Wno-unused-result \
-pipe
override CXXFLAGS += \
-std=c++17 \
-std=c++20 \
-MMD \
-MP
override INC_FLAGS := \
@@ -125,8 +131,7 @@ override INC_FLAGS := \
-Ivendored \
-Ivendored/libfuse/include
override MFS_FLAGS := \
-DUSE_XATTR=$(USE_XATTR) \
-DUGID_USE_RWLOCK=$(UGID_USE_RWLOCK)
-DUSE_XATTR=$(USE_XATTR)
override TESTS_FLAGS := \
-Isrc \
-Ivendored \
@@ -198,7 +203,7 @@ libfuse: $(LIBFUSE)
$(MAKE) -C vendored/libfuse
$(LIBFUSE):
$(MAKE) -C vendored/libfuse $(BUILDDIR)/libfuse.a
$(MAKE) -C vendored/libfuse $(BUILDDIR)/libfuse.a RELEASE=$(RELEASE)
tests: $(BUILDDIR)/tests
@@ -340,7 +345,7 @@ rpm: tarball
$(SED) 's/__VERSION__/$(VERSION)/g' mergerfs.spec > \
rpmbuild/SOURCES/mergerfs.spec
cp -ar mergerfs-$(VERSION).tar.gz rpmbuild/SOURCES
rpmbuild -ba rpmbuild/SOURCES/mergerfs.spec \
$(RPMBUILD) $(RPMBUILD_FLAGS) -ba rpmbuild/SOURCES/mergerfs.spec \
--define "_topdir $(CURDIR)/rpmbuild"
.PHONY: install-build-pkgs
@@ -362,14 +367,16 @@ define build_release
--branch=$(GITREF)
endef
.PHONY: release
.PHONY: release release-all
.PHONY: release-amd64 release-arm64 release-armhf release-riscv64
.PHONY: release-sample release-static release-tarball
.PHONY: release-sample release-static release-tarball release-installer
release:
$(MAKE) RELEASE=1 CPPFLAGS="$(CPPFLAGS) -DNDEBUG" all
release-all:
$(call build_release,"all")
release-sample:
$(if $(RELEASE),,$(eval RELEASE := $(shell ./buildtools/detect-local-target 2>/dev/null)))
$(call build_release,$(if $(RELEASE),$(RELEASE),$(DEFAULT_RELEASE)))
$(if $(TARGET),,$(eval TARGET := $(shell ./buildtools/detect-local-target 2>/dev/null)))
$(call build_release,$(if $(TARGET),$(TARGET),$(DEFAULT_TARGET)))
release-amd64:
$(call build_release,"amd64")
release-arm64:
@@ -382,6 +389,8 @@ release-static:
$(call build_release,"static")
release-tarball:
$(call build_release,"tarball")
release-installer:
$(call build_release,"installer")
container:
$(eval GITREF ?= $(shell git describe --exact-match --tags HEAD 2>/dev/null || git symbolic-ref --short HEAD 2>/dev/null || git rev-parse --short HEAD))
+15 -3
View File
@@ -1,6 +1,19 @@
#!/bin/sh
set -e
if [ -e /tmp/build-env ]; then
. /tmp/build-env
fi
rpmbuild_flags()
{
case "$(command -v "${CXX:-g++}")" in
/opt/rh/*)
printf '%s' '--undefine _annotated_build'
;;
esac
}
if [ $# -ne 2 ]; then
echo "usage: $0 <branch> <git_repo>"
echo "argv: $@"
@@ -28,10 +41,9 @@ mkdir "/build"
if [ -e /usr/bin/apt-get ]; then
make deb
elif [ -e /usr/bin/dnf ]; then
make rpm
make RPMBUILD_FLAGS="$(rpmbuild_flags)" rpm
elif [ -e /usr/bin/yum ]; then
. /opt/rh/devtoolset-9/enable
make rpm
make RPMBUILD_FLAGS="$(rpmbuild_flags)" rpm
elif [ -e /sbin/apk ]; then
echo "NOT YET SUPPORTED"
exit 1
+1 -1
View File
@@ -18,7 +18,7 @@ def build(git_repo,
'--force-rm']
if os.path.exists(git_repo):
git_repo = os.path.realpath(git_repo)
args += ['-v',f'{git_repo}:/tmp/mergerfs.git:ro']
args += ['-v',f'{git_repo}:/tmp/mergerfs.git:ro:z']
args += ['--build-arg=GIT_REPO=file:///tmp/mergerfs.git']
else:
args += [f'--build-arg=GIT_REPO={git_repo}']
+1 -1
View File
@@ -1,4 +1,4 @@
FROM --platform=amd64 fedora:43 as build
FROM --platform=amd64 fedora:44 as build
COPY install-build-pkgs build-mergerfs /tmp/
RUN /tmp/install-build-pkgs
ARG BUILD_TIMESTAMP=0
+1 -1
View File
@@ -1,4 +1,4 @@
FROM --platform=arm64 fedora:43 as build
FROM --platform=arm64 fedora:44 as build
COPY install-build-pkgs build-mergerfs /tmp/
RUN /tmp/install-build-pkgs
ARG BUILD_TIMESTAMP=0
+4 -1
View File
@@ -1,6 +1,9 @@
ARG BUILD_TIMESTAMP=0
FROM --platform=linux/amd64 ubuntu:latest as build
RUN apt-get update && apt-get install -y makeself
COPY install-build-pkgs /tmp/
RUN /tmp/install-build-pkgs && \
DEBIAN_FRONTEND=noninteractive apt-get -qy --no-install-suggests --no-install-recommends install \
makeself
RUN mkdir -p /tmp/mergerfs/
RUN ls -lhd /pkgs
RUN cp -v /pkgs/*static*.gz /tmp/mergerfs/
+126 -3
View File
@@ -1,39 +1,162 @@
#!/bin/sh
set -e
BUILD_ENV=/tmp/build-env
write_gcc_env()
{
{
printf 'export CC=%s\n' "$1"
printf 'export CXX=%s\n' "$2"
} > "$BUILD_ENV"
}
write_toolset_env()
{
{
printf '. %s\n' "$1"
printf 'export CC=gcc\n'
printf 'export CXX=g++\n'
} > "$BUILD_ENV"
}
latest_apt_gxx()
{
apt-cache --names-only search '^g[+][+]-[0-9]' \
| sed -n 's/^\(g++-[0-9][0-9]*\) .*/\1/p' \
| sort -t- -k2,2V \
| tail -n 1
}
latest_dnf_toolset()
{
dnf -q list --showduplicates 'gcc-toolset-*-gcc-c++' 2>/dev/null \
| sed -n 's/^gcc-toolset-\([0-9][0-9]*\)-gcc-c++\..*/\1/p' \
| sort -V \
| tail -n 1
}
latest_yum_toolset()
{
yum -q list available 'devtoolset-*-gcc-c++' 2>/dev/null \
| sed -n 's/^devtoolset-\([0-9][0-9]*\)-gcc-c++\..*/\1/p' \
| sort -V \
| tail -n 1
}
version_gt()
{
highest="$(printf '%s\n%s\n' "$1" "$2" | sort -V | tail -n 1)"
[ "$highest" = "$1" ] && [ "$1" != "$2" ]
}
check_cxx20()
{
cxx="${CXX:-g++}"
test_src=/tmp/mergerfs-cxx20-check.cpp
test_bin=/tmp/mergerfs-cxx20-check
cat > "$test_src" <<'EOF'
template <typename T>
concept Incrementable = requires(T value) { value + 1; };
static_assert(Incrementable<int>);
int main()
{
return 0;
}
EOF
if ! "$cxx" -std=c++20 "$test_src" -o "$test_bin" >/dev/null 2>&1; then
echo "error: $cxx does not support C++20" >&2
exit 1
fi
rm -f "$test_src" "$test_bin"
}
: > "$BUILD_ENV"
if [ -e /usr/bin/apt-get ]; then
export DEBIAN_FRONTEND=noninteractive
apt-get -qy update
cxx_pkg="$(latest_apt_gxx)"
if [ -n "$cxx_pkg" ]; then
gcc_pkg="gcc-${cxx_pkg#g++-}"
compiler_pkgs="$gcc_pkg $cxx_pkg"
write_gcc_env "$gcc_pkg" "$cxx_pkg"
else
compiler_pkgs="g++"
write_gcc_env gcc g++
fi
apt-get -qy --force-yes --no-install-suggests --no-install-recommends \
install \
ca-certificates \
build-essential \
dpkg-dev \
git \
g++ \
$compiler_pkgs \
debhelper \
lsb-release \
fakeroot
. "$BUILD_ENV"
check_cxx20
elif [ -e /usr/bin/dnf ]; then
dnf -y update
toolset="$(latest_dnf_toolset)"
dnf -y install \
git rpm-build gcc-c++ make which libatomic
default_gxx_major="$(g++ -dumpversion | cut -f1 -d.)"
if [ -n "$toolset" ] && version_gt "$toolset" "$default_gxx_major"; then
dnf -y install \
"gcc-toolset-${toolset}-gcc" \
"gcc-toolset-${toolset}-gcc-c++"
write_toolset_env "/opt/rh/gcc-toolset-${toolset}/enable"
else
write_gcc_env gcc g++
fi
. "$BUILD_ENV"
check_cxx20
elif [ -e /usr/bin/yum ]; then
yum -y update
yum -y install \
git rpm-build gcc-c++ make which
yum -y install centos-release-scl
yum -y install devtoolset-9-gcc\*
yum -y install centos-release-scl || true
yum -y update
toolset="$(latest_yum_toolset)"
default_gxx_major="$(g++ -dumpversion | cut -f1 -d.)"
if [ -n "$toolset" ] && version_gt "$toolset" "$default_gxx_major"; then
yum -y install \
"devtoolset-${toolset}-gcc" \
"devtoolset-${toolset}-gcc-c++"
write_toolset_env "/opt/rh/devtoolset-${toolset}/enable"
else
write_gcc_env gcc g++
fi
. "$BUILD_ENV"
check_cxx20
elif [ -e /usr/bin/zypper ]; then
zypper update -y
zypper install -y \
git rpm-build gcc-c++ make which
write_gcc_env gcc g++
. "$BUILD_ENV"
check_cxx20
elif [ -e /sbin/apk ]; then
apk add \
abuild git gcc g++ make \
linux-headers
write_gcc_env gcc g++
. "$BUILD_ENV"
check_cxx20
elif [ -e /usr/sbin/pkg ]; then
pkg install \
git gmake gcc
elif [ -e /usr/bin/pacman ]; then
pacman -Sy --noconfirm base-devel git fakeroot
write_gcc_env gcc g++
. "$BUILD_ENV"
check_cxx20
fi
+4 -1
View File
@@ -6,8 +6,11 @@ export DH_VERBOSE=1
%:
dh $@
override_dh_auto_build:
$(MAKE) release
override_dh_auto_install:
$(MAKE) \
DESTDIR=$(CURDIR)/debian/mergerfs \
PREFIX=/usr \
install
install
+3
View File
@@ -155,6 +155,9 @@ config file.
multiplied by the number of process threads plus read thread
count. 0 sets the depth to the same as the process thread
count. (default: 2)
* **[fuse-read-spin-us](threads.md)=UINT**: Microseconds a read thread
polls for the next request before sleeping. Trades CPU for lower
latency on synchronous workloads. (default: 0)
* **[pin-threads](pin-threads.md)=STR**: Selects a strategy to pin
threads to CPUs (default: unset)
* **[flush-on-close](flush-on-close.md)=never|always|opened-for-write**:
+23
View File
@@ -59,3 +59,26 @@ in the pool.
block in order to limit memory growth.
* `process-thread-queue-depth<=0`: Sets the queue depth to 2. May be
used in the future to set dynamically.
## fuse-read-spin-us
Defaults to `0` (disabled)
Applies only when `process-thread-count=-1` (read threads also process
requests). After handling a request, each read thread polls for the
next request for up to this many microseconds before going to sleep.
This lowers per-request latency for synchronous, single-stream workloads
(sequential reads, small writes, metadata-heavy walks such as `ls -l`,
`find`, or `rsync` scans) at the cost of CPU: while requests keep
arriving, a read thread stays busy instead of sleeping. On a quiet,
low-rate request stream this costs roughly one CPU core's worth of
polling with no benefit. An idle mount costs nothing.
`50` is the only value tested. Larger values increase CPU use. Parallel
multi-stream workloads and large (1MiB) writes showed no change.
* `fuse-read-spin-us=0`: Disabled. Read threads block immediately.
* `fuse-read-spin-us=N`: Poll for up to `N` microseconds after each
request.
@@ -149,6 +149,70 @@ It is generally recommended to not use user namespacing / id mapping
given the complication it introduces.
## Does mergerfs work with SELinux relabeling?
FUSE filesystems, including mergerfs, do not support per-inode
SELinux labels (see [Red Hat bug
1272868](https://bugzilla.redhat.com/show_bug.cgi?id=1272868) and the
[upstream fuse-devel
thread](https://sourceforge.net/p/fuse/mailman/fuse-devel/thread/CAN9vWDLGA4%3Dg3peKVWkX7gLRiVUUvJAQx9kiXHgqdeX6eh0MYA%40mail.gmail.com/)).
When a container runtime such as Podman or Docker is asked to relabel
a bind mount with the `:z` or `:Z` mount option, it first probes the
source for `security.selinux` xattr support. On a FUSE path that
probe returns `EOPNOTSUPP` and the runtime skips relabeling entirely.
Files inside the mount keep the `fusefs_t` type assigned by the
kernel. No tree walk is performed and no journal noise is produced at
default verbosity. The runtime logs `Labeling not supported on
"<path>"` only at debug level (for example
`podman --log-level=debug`). The flag is therefore a no-op against a
mergerfs mount; not harmful, but misleading.
Containers can still read and write a mergerfs pool without `:z`
because two independent access checks must succeed and both can be
satisfied without any relabel:
* **SELinux:** on Fedora, RHEL / CentOS Stream, and SUSE the shipped
[container-selinux](https://github.com/containers/container-selinux)
policy already permits the `container_t` process type to manage
files of type `fusefs_t`. See `container_selinux(8)`.
* **DAC:** the in-container UID and GID must still be able to read
and write the files on the underlying branches. With rootless
Podman this is most easily arranged by aligning the container UID
with the host owner of the branches (for example
`--userns=keep-id`, or Quadlet `UserNS=keep-id`).
The recommendation is to omit `:z` and `:Z` on bind mounts whose
source is a mergerfs (or any other FUSE) path, and to keep them on
bind mounts whose source is a regular local filesystem (ext4, btrfs,
xfs, etc.) where relabeling does real work. On distributions whose
container-selinux policy does not already include the `fusefs_t`
allow rule, the targeted fix is to set a single label at mount time
by adding `context="system_u:object_r:container_file_t:s0"` to the
mergerfs mount options (for example in `/etc/fstab`). Files inside
the pool then appear as `container_file_t`, which container-selinux
permits `container_t` to manage everywhere. See [Red Hat's mounting
file systems
guide](https://docs.redhat.com/en/documentation/red_hat_enterprise_linux/7/html/selinux_users_and_administrators_guide/sect-security-enhanced_linux-working_with_selinux-mounting_file_systems).
This is preferred over the much broader
`--security-opt label=disable`, which removes SELinux confinement
from the container as a whole. The trade-off of `context=` is that
host-side processes touching the mount (rsync, backup tools, web
UIs, cron jobs) see `container_file_t` rather than `fusefs_t` and
may require a small SELinux policy adjustment.
To verify on your own system:
```
podman --log-level=debug run --rm \
-v /path/to/mergerfs:/data:z \
alpine true 2>&1 | grep -iE 'relabel|setfilecon|Labeling'
```
Expected output is a single line of the form
`Labeling not supported on "/path/to/mergerfs"` and nothing in
`ausearch -m AVC,SELINUX_ERR` for the mount.
## Can mergerfs be used with NFS root squash?
If mergerfs is pooling a NFS mount then root squash must be disabled
@@ -215,13 +215,63 @@ If `mergerfs` doesn't work as a type it could be due to how the
permissions.
## Why was splice support removed?
## Does mergerfs use splice?
After a lot of testing over the years, splicing always appeared to
at best, provide equivalent performance, and in some cases, worse
performance. Splice is not supported on other platforms forcing a
traditional read/write fallback to be provided. The splice code was
removed to simplify the codebase.
On Linux, for large reads only, automatically. When a `read` request
is 64KiB or larger, mergerfs splices the data from the branch file
through a pipe into `/dev/fuse` instead of reading it into its own
memory and writing it back out. The data never crosses userspace, which
saves one copy and some CPU per request. Smaller reads, all writes and
every other request use the normal copy path, because splicing them
costs more in extra syscalls than it saves: earlier versions that
spliced every request were measurably slower for small I/O and metadata
operations.
There is nothing to configure. If splicing a reply is not possible the
read is served through the copy path instead: a branch filesystem that
does not support splice is detected on the first attempt and skipped for
that open file, and a read extending past the size the branch reports
for the file (the last part of a file, files smaller than the read,
files reporting a size of 0 such as procfs files, and files reporting a
nominal 4096 bytes such as sysfs attributes) uses the copy path, so a
branch whose reported size lags its data still
returns everything. Files on FUSE, SMB/CIFS, NFS, Ceph, AFS, overlayfs
or 9p branches, and any file opened `O_DIRECT`, always use the copy
path: those filesystems' splice_read can serve a page cache that their
normal reads bypass or first revalidate, so a spliced reply could
return stale data, and for an `O_DIRECT` file the kernel would serve
the splice by direct I/O into newly allocated pages, which saves
nothing over the copy path.
A read that arrives while mergerfs is moving the file to another
branch (`moveonenospc`) also uses the copy path.
Unless mergerfs runs with
`CAP_SYS_RESOURCE` (as root on a host normally does, but root in a
container usually does not unless started with `--cap-add=SYS_RESOURCE`
or `--privileged`; `CAP_SYS_ADMIN` is not enough), the `pipe-max-size`
sysctl limits the pipe a reply can use. A reply needs one page more
than the read (for the reply header), two when the read does not start
on a page boundary, and the kernel rounds that up to a power of two, so
`pipe-max-size` must be larger than `fuse-msg-size` for the largest
reads to be spliced: with the defaults (both 1MiB) 1MiB reads use the
copy path while smaller reads keep splicing, with a single log
message. That limit is read when mergerfs starts, so raising
`pipe-max-size` takes effect after mergerfs is restarted. Only a kernel
without splice support, or a seccomp policy that refuses it, disables it
for the whole process. The old
`splice_read`, `splice_write` and `splice_move` options are still
accepted and do nothing.
Two behaviors of spliced reads differ from the copy path. The data is
copied into the reply only after the branch filesystem's read has
returned, so a write or truncate of the branch file racing a read can
produce a reply mixing old and new contents, even on filesystems such as
XFS that otherwise serialize reads against writes. And on kernels before
6.4, spliced reads do not generate inotify or fanotify access events
(`IN_ACCESS`, `FAN_ACCESS`) on the branch file.
Files using [passthrough.io](../config/passthrough.md) never reach
mergerfs for reads, so splice does not apply to them, unless
`direct-io` is also in effect for them, which overrides passthrough.
## How does mergerfs handle credentials?
+59
View File
@@ -48,6 +48,65 @@ before changing them to understand how functionality will change.
* use LVM and LVM cache to place a SSD in front of your HDDs
## Understanding mergerfs IO Performance
A common source of confusion when evaluating **mergerfs**' performance
is comparing it to traditional RAID or filesystem level pooling
technologies such as mdadm, ZFS, btrfs, or striped LVM. Those
solutions aggregate devices into a single virtual device and spread IO
across all members. Under favorable conditions that means the
performance of the pool is the sum of the performance of its members:
striped reads and writes hit multiple drives at once and the aggregate
throughput scales with the number of devices, up to the limits of the
interconnects (PCIe, SATA, network, etc.) which every member must
share.
**mergerfs** does not aggregate or stripe data. Every file lives
wholly on one branch and **mergerfs** simply routes access to it. As a
result **mergerfs** does not aggregate the performance of its branches
in the way RAID does. What it provides instead is **independent
performance across branches**.
* **Best case:** workloads which access different branches
concurrently perform independently of one another. If you have 3
filesystems on 3 drives and each is being actively used, the total
throughput can be A + B + C: the sum of each branch performing at
full speed. No single drive is a bottleneck for the others.
* **Worst case:** a single stream of IO - one process reading or
writing one file - performs at the speed of the single branch the
file is on, no differently than a lone filesystem. Additional
branches do not make a single file faster.
This means that when reasoning about expected performance the key
question is whether your workload is **single stream** or **multi
stream**:
* Single large file transfers, video playback, or one process doing
sequential IO will never exceed the speed of one branch regardless
of how many are mounted. If a single stream must be faster, put a
faster device (tiered cache, SSD, LVM cache) in front of the
branches or use an actual striped RAID layer beneath **mergerfs**'
branches.
* Concurrency across branches is where **mergerfs** shines. Many
parallel downloads, seeding torrents, multiple rsync streams,
multiple applications reading different files: these naturally
spread across branches and their throughput adds up.
There are practical upsides to this non-aggregated design as well:
* Drives of different sizes and speeds don't reduce the whole pool to
the slowest member. Each branch performs at its own speed.
* The failure of one drive only affects the files on that branch
rather than the integrity of the entire pool.
Finally, keep in mind that **mergerfs** is a proxy above the branches,
so measured throughput is the lesser of the branch's native performance
and any overhead introduced by **mergerfs** and FUSE. Use
[nullrw](config/options.md) or a ram disk as a branch to measure that
overhead separately, and see the [benchmarking
section](benchmarking.md) for how to benchmark meaningfully.
## Additional Reading
* [Benchmarking](benchmarking.md)
+23 -1
View File
@@ -135,11 +135,33 @@ docker pull ghcr.io/trapexit/mergerfs:TAG
docker run --device=/dev/fuse --cap-add=SYS_ADMIN -v /mnt/to-merge:/mnt/to-merge:rshared -v /mnt/mergerfs:/mnt/mergerfs:z,shared ghcr.io/trapexit/mergerfs:TAG -f '/mnt/to-merge/*' '/mnt/mergerfs'
```
or with docker compose
```
services:
mergerfs:
image: ghcr.io/trapexit/mergerfs:TAG
container_name: mergerfs
restart: unless-stopped
privileged: true
cap_add:
- SYS_ADMIN
devices:
- /dev/fuse:/dev/fuse
command: -f '/mnt/to-merge/*' '/mnt/mergerfs'
volumes:
- /mnt/to-merge:/mnt/to-merge:rshared
- /mnt/mergerfs:/mnt/mergerfs:z,shared
```
* `--device=/dev/fuse`: Pass in host FUSE device.
* `--cap-add=SYS_ADMIN`: Give proper permissions to mount.
* `-v /mnt/mergerfs:/mnt/mergerfs:z,shared`: `z` and `shared` will
allow the mount to be shared between the container and host and
therefore other containers.
therefore other containers. Note that once mergerfs is mounted
inside the container the resulting FUSE filesystem does not
support SELinux relabeling. See [Does mergerfs work with SELinux
relabeling?](../faq/compatibility_and_integration.md#does-mergerfs-work-with-selinux-relabeling)
* `-v /mnt/to-merge:/mnt/to-merge:rshared`: The mounts from the host
to merge within the container.
+68 -28
View File
@@ -28,8 +28,10 @@
#include "str.hpp"
#include "syslog.hpp"
#include <string>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <fnmatch.h>
@@ -437,24 +439,44 @@ Branches::from_string(const std::string_view str_)
Branches::Ptr impl;
Branches::Ptr new_impl;
impl = std::atomic_load(&_impl);
new_impl = std::make_shared<Branches::Impl>(&minfreespace);
*new_impl = *impl;
rv = new_impl->from_string(str_);
if(rv < 0)
return rv;
for(;;)
{
{
std::shared_lock<std::shared_mutex> lk(_mutex);
impl = _impl;
}
std::atomic_store(&_impl,new_impl);
*new_impl = *impl;
return 0;
rv = new_impl->from_string(str_);
if(rv < 0)
return rv;
{
std::unique_lock<std::shared_mutex> lk(_mutex);
if(impl != _impl)
continue;
_impl = std::move(new_impl);
}
return 0;
}
}
std::string
Branches::to_string(void) const
{
return std::atomic_load(&_impl)->to_string();
Branches::Ptr impl;
{
std::shared_lock<std::shared_mutex> lk(_mutex);
impl = _impl;
}
return impl->to_string();
}
void
@@ -462,32 +484,50 @@ Branches::find_and_set_mode_ro()
{
Branches::Ptr impl;
Branches::Ptr new_impl;
std::vector<fs::path> ro_paths;
impl = std::atomic_load(&_impl);
new_impl = std::make_shared<Branches::Impl>(&minfreespace);
*new_impl = *impl;
bool changed = false;
for(auto &branch : *new_impl)
for(;;)
{
if(branch.mode != Branch::Mode::RW)
continue;
ro_paths.clear();
if(!fs::is_rofs_but_not_mounted_ro(branch.path))
continue;
{
std::shared_lock<std::shared_mutex> lk(_mutex);
impl = _impl;
}
SysLog::warning("branch `{}` found to be readonly - setting its mode=RO",
branch.path.string());
*new_impl = *impl;
branch.mode = Branch::Mode::RO;
changed = true;
for(auto &branch : *new_impl)
{
if(branch.mode != Branch::Mode::RW)
continue;
if(!fs::is_rofs_but_not_mounted_ro(branch.path))
continue;
branch.mode = Branch::Mode::RO;
ro_paths.emplace_back(branch.path);
}
if(ro_paths.empty())
return;
{
std::unique_lock<std::shared_mutex> lk(_mutex);
if(impl != _impl)
continue;
_impl = std::move(new_impl);
}
for(const auto &path : ro_paths)
SysLog::warning("branch `{}` found to be readonly - setting its mode=RO",
path.string());
return;
}
if(not changed)
return;
std::atomic_store(&_impl,new_impl);
}
SrcMounts::SrcMounts(Branches &b_)
+14 -4
View File
@@ -23,9 +23,9 @@
#include "strvec.hpp"
#include "tofrom_string.hpp"
#include <atomic>
#include <cstdint>
#include <memory>
#include <shared_mutex>
#include <string>
#include <vector>
@@ -69,7 +69,8 @@ public:
u64 minfreespace = MINFREESPACE_DEFAULT;
private:
mutable std::shared_ptr<Impl> _impl;
Branches::Ptr _impl;
mutable std::shared_mutex _mutex;
public:
Branches()
@@ -82,8 +83,17 @@ public:
std::string to_string(void) const final;
public:
operator Ptr() const { return std::atomic_load(&_impl); }
Ptr operator->() const { return std::atomic_load(&_impl); }
operator Ptr() const
{
std::shared_lock<std::shared_mutex> lk(_mutex);
return _impl;
}
Ptr operator->() const
{
std::shared_lock<std::shared_mutex> lk(_mutex);
return _impl;
}
public:
void find_and_set_mode_ro();
+3
View File
@@ -116,6 +116,7 @@ Config::Config()
posix_acl(false),
process_thread_count(fuse_cfg.process_thread_count),
process_thread_queue_depth(fuse_cfg.process_thread_queue_depth),
fuse_read_spin_us(fuse_cfg.read_spin_us),
proxy_ioprio(false),
read_thread_count(fuse_cfg.read_thread_count),
readahead(0),
@@ -172,6 +173,7 @@ Config::Config()
posix_acl.ro =
process_thread_count.ro =
process_thread_queue_depth.ro =
fuse_read_spin_us.ro =
read_thread_count.ro =
scheduling_priority.ro =
true;
@@ -276,6 +278,7 @@ Config::Config()
_map["posix-acl"] = &posix_acl;
_map["process-thread-count"] = &process_thread_count;
_map["process-thread-queue-depth"] = &process_thread_queue_depth;
_map["fuse-read-spin-us"] = &fuse_read_spin_us;
_map["proxy-ioprio"] = &proxy_ioprio;
_map["read-thread-count"] = &read_thread_count;
_map["readahead"] = &readahead;
+1
View File
@@ -153,6 +153,7 @@ public:
ConfigBOOL posix_acl;
TFSRef<int> process_thread_count;
TFSRef<int> process_thread_queue_depth;
TFSRef<u64> fuse_read_spin_us;
ProxyIOPrio proxy_ioprio;
TFSRef<int> read_thread_count;
ConfigU64 readahead;
+92 -3
View File
@@ -48,10 +48,12 @@
#include "fh.hpp"
#include "fs_path.hpp"
#include "mutex.hpp"
#include "base_types.h"
#include <atomic>
#include <mutex>
#include <shared_mutex>
class FileInfo : public FH
{
@@ -97,7 +99,94 @@ public:
int fd;
Branch branch;
u32 direct_io:1;
Mutex mutex;
// Serializes the fd state across concurrent writes on the same open
// file. Concurrent writes happen with:
// 1) writeback-cache + page-cache mode
// 2) parallel_direct_writes + direct_io=true
//
// Common case (plain pwrite on stable fd) only needs a shared lock
// since the kernel already serializes fd mutations. The danger is
// the fs::dup2() rebinding when moveonenospc activates. Writers
// take shared lock; move takes unique (lock_exclusive). Spliced
// reads (FUSE::read_buf) also take it shared, but only if that
// needs no waiting.
std::shared_mutex mutex;
// Threads waiting in lock_exclusive. FUSE::read_buf starts no splice
// while one waits: the rwlock prefers readers, so overlapping spliced
// reads could otherwise hold a mover off indefinitely.
std::atomic<int> exclusive_waiters{0};
// Lower bound on the branch file's size, used only by FUSE::read_buf
// to skip its end-of-file size check for reads ending inside this
// bound. Raised by that check, lowered by any short read (FUSE::read)
// or ftruncate and reset by fallocate through this handle. It may go
// stale when the file shrinks through another handle or outside
// mergerfs; the splice reply then hits EOF and falls back to
// FUSE::read, which returns the correct bytes and lowers the bound
// to where that read found the end (to 0 when it read nothing, as
// the end is then somewhere at or before its offset). 0 = unknown.
std::atomic<s64> size_floor{0};
// Whether FUSE::read_buf may offer fd for splicing: unknown until the
// first large read checks the branch filesystem (some serve splice
// from a page cache their reads bypass or revalidate), no once that
// check said so or the fd refused splice (its filesystem has no
// splice_read), after which reads go straight to the copy path.
// Decided once per fd binding; a filesystem refused by type is also
// remembered by device, so later handles on it are settled without
// repeating that check (see FUSE::read_buf).
enum class Splice : uint8_t { unknown, yes, no };
std::atomic<Splice> splice_state{Splice::unknown};
// Take mutex exclusively (moveonenospc), announced through
// exclusive_waiters so that no new spliced read starts meanwhile.
std::unique_lock<std::shared_mutex>
lock_exclusive()
{
// Counted out again even if locking throws, so a failed attempt
// cannot leave spliced reads on this handle declined for good.
struct Waiting
{
std::atomic<int> &count;
explicit Waiting(std::atomic<int> &count_)
: count(count_)
{
count.fetch_add(1,std::memory_order_relaxed);
}
~Waiting()
{
count.fetch_sub(1,std::memory_order_relaxed);
}
} waiting(exclusive_waiters);
return std::unique_lock<std::shared_mutex>(mutex);
}
void size_floor_lower(const s64 v_)
{
s64 cur = size_floor.load(std::memory_order_relaxed);
while((v_ < cur) &&
!size_floor.compare_exchange_weak(cur,v_,std::memory_order_relaxed))
;
}
// The file's size may have changed in a way size_floor_lower cannot
// express: call after the change. That narrows, but does not close,
// the window in which a size check that started before the change
// (and saw 0, the value stored here) installs the old size
// afterwards; like any stale floor that only costs a fallback.
void size_floor_reset()
{
size_floor.store(0,std::memory_order_relaxed);
}
// fd now refers to a different branch file (moveonenospc's dup2): a
// splice decision or size observed on the old one no longer applies.
void fd_rebound()
{
splice_state.store(Splice::unknown,std::memory_order_relaxed);
size_floor_reset();
}
};
inline
+24 -16
View File
@@ -21,8 +21,8 @@
#include "fs_statvfs.hpp"
#include "statvfs_util.hpp"
#include "mutex.hpp"
#include <mutex>
#include <shared_mutex>
#include <string>
#include <unordered_map>
@@ -38,9 +38,11 @@ struct Element
typedef std::unordered_map<std::string,Element> statvfs_cache;
static u64 g_timeout = 0;
static statvfs_cache g_cache;
static Mutex g_cache_lock;
// Cache is read heavy. Use a shared mutex so concurrent reads can
// work in parallel and only inserts or refreshes take a unique lock.
static std::shared_mutex g_cache_mutex;
static u64 g_timeout = 0;
static statvfs_cache g_cache;
static
@@ -70,29 +72,35 @@ int
fs::statvfs_cache(const std::string &path_,
struct statvfs *st_)
{
int rv;
Element *e;
u64 now;
if(g_timeout == 0)
return fs::statvfs(path_,st_);
rv = 0;
now = ::_get_time();
mutex_lock(g_cache_lock);
{
std::shared_lock<std::shared_mutex> lk(g_cache_mutex);
e = &g_cache[path_];
auto it = g_cache.find(path_);
if((it != g_cache.end()) && ((now - it->second.time) <= g_timeout))
{
*st_ = it->second.st;
return 0;
}
}
if((now - e->time) > g_timeout)
int rv = 0;
std::unique_lock<std::shared_mutex> lk(g_cache_mutex);
Element &e = g_cache[path_];
if((now - e.time) > g_timeout)
{
e->time = now;
rv = fs::statvfs(path_,&e->st);
e.time = now;
rv = fs::statvfs(path_,&e.st);
}
*st_ = e->st;
mutex_unlock(g_cache_lock);
*st_ = e.st;
return rv;
}
+1 -1
View File
@@ -334,7 +334,7 @@ _create(const fuse_req_ctx_t *ctx_,
delete fi;
ffi_->fh = 0;
SysLog::crit(msg);
SysLog::crit("{}",msg);
fmt::println(stderr,"{}",msg);
return -EIO;
+13 -4
View File
@@ -55,14 +55,23 @@ FUSE::fallocate(const fuse_req_ctx_t *ctx_,
off_t offset_,
off_t len_)
{
int rv;
FileInfo *fi;
fi = state.get_fi(ctx_,fh_);
if(not fi)
return -EBADF;
return ::_fallocate(fi->fd,
mode_,
offset_,
len_);
rv = ::_fallocate(fi->fd,
mode_,
offset_,
len_);
/* Reset read_buf's size floor. Defensive only: of the fallocate
modes only COLLAPSE_RANGE shrinks a file (PUNCH_HOLE requires
KEEP_SIZE), and the kernel's FUSE client forwards only KEEP_SIZE,
PUNCH_HOLE and ZERO_RANGE, so no request reaching here does. */
fi->size_floor_reset();
return rv;
}
+10 -1
View File
@@ -50,11 +50,20 @@ FUSE::ftruncate(const fuse_req_ctx_t *ctx_,
cu64 fh_,
off_t size_)
{
int rv;
FileInfo *fi;
fi = state.get_fi(ctx_,fh_);
if(not fi)
return -EBADF;
return ::_ftruncate(fi->fd,size_);
rv = ::_ftruncate(fi->fd,size_);
/* A shrink through this handle would otherwise leave reads past the
new end on the EOF fallback; after a grow the old floor still
holds. */
if(rv == 0)
fi->size_floor_lower(size_);
return rv;
}
+10 -1
View File
@@ -367,10 +367,19 @@ _open(const fuse_req_ctx_t *ctx_,
// The ref increment keeps fuse_release.cpp from erasing and
// freeing the canonical entry while we duplicate its fd.
//
// memory_order_relaxed is sufficient for the increment. The
// concurrent_flat_map bucket lock held inside visit() already
// provides the happens-before relationship with other mutators
// of the same key (the release() erase_if() callback and other
// open() visitors.) The lock's release/acquire semantics
// publish the ref_count update to subsequent observers of the
// same entry, so we only need atomicity from the fetch_add
// itself.
of.visit(ctx_->nodeid,
[&](auto &v_)
{
v_.second.ref_count++;
v_.second.ref_count.fetch_add(1,std::memory_order_relaxed);
fi = v_.second.fi;
backing_id = v_.second.backing_id;
});
+12 -3
View File
@@ -72,10 +72,19 @@ FUSE::read(const fuse_req_ctx_t *ctx_,
if(not fi)
return -EBADF;
if(fi->direct_io)
return ::_read_direct_io(fi->fd,buf_,size_,offset_);
const int rv = (fi->direct_io ?
::_read_direct_io(fi->fd,buf_,size_,offset_) :
::_read_cached(fi->fd,buf_,size_,offset_));
return ::_read_cached(fi->fd,buf_,size_,offset_);
/* A short read means end-of-file at offset+rv: keep read_buf's size
floor from claiming more (it is the splice reply's EOF fallback).
Nothing was read at all when rv is 0, so the file may end anywhere
at or before offset_ (it may have shrunk well below the offset
asked about): only 0 is then a bound that cannot be too high. */
if((rv >= 0) && ((size_t)rv < size_))
fi->size_floor_lower((rv > 0) ? ((s64)offset_ + rv) : 0);
return rv;
}
int
+313
View File
@@ -0,0 +1,313 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include "fuse_read_buf.hpp"
#include "fileinfo.hpp"
#include "fs_fstat.hpp"
#include "fs_getfl.hpp"
#include "fs_statx.hpp"
#include "ioprio.hpp"
#include "state.hpp"
#include "fuse.h"
#include "boost/unordered/concurrent_flat_map.hpp"
#include <atomic>
#include <fcntl.h>
#include <shared_mutex>
#if defined(__linux__)
#include <sys/sysmacros.h>
#include <sys/vfs.h>
#endif
/* The branch file's type, size and device. A statx asking only for the
first two lets filesystems skip work a full fstat implies (NFS
flushes the file's dirty pages before reporting c/mtime; the device
is always reported); fstat covers builds and filesystems without
statx. */
static
int
_type_and_size(const int fd_,
mode_t *mode_,
s64 *size_,
dev_t *dev_)
{
int rv;
struct stat st;
#ifdef MERGERFS_SUPPORTED_STATX
struct fuse_statx stx;
const unsigned int want = (STATX_TYPE|STATX_SIZE);
rv = fs::statx(fd_,"",AT_EMPTY_PATH,want,&stx);
if((rv == 0) && ((stx.mask & want) == want))
{
*mode_ = stx.mode;
*size_ = (s64)stx.size;
*dev_ = makedev(stx.dev_major,stx.dev_minor);
return 0;
}
#endif
rv = fs::fstat(fd_,&st);
if(rv < 0)
return rv;
*mode_ = st.st_mode;
*size_ = st.st_size;
*dev_ = st.st_dev;
return 0;
}
/* Devices of branch filesystems _splice_coherent refused by type. The
type is a property of the superblock, not of the open file, and on
the network and FUSE filesystems it refuses fstatfs is a round trip
to their server, which the first large read of every opened file
would otherwise pay. Only refusals are remembered, so a device
number reused by another filesystem after an unmount can cost
splicing but never coherence. The value is unused. */
static boost::concurrent_flat_map<dev_t,bool> g_refused_devs;
/* Whether splicing from the branch fd returns what pread would, and is
worth doing. A filesystem that cannot be identified is refused too.
Not on FUSE, SMB/CIFS, NFS, Ceph, AFS, overlayfs or 9p
branches: their splice_read serves the branch inode's page cache
(filemap_splice_read) while their read path bypasses it or first
revalidates it (FUSE direct_io and auto_inval_data, SMB without a
read lease, NFS/Ceph/AFS/9p attribute/data caching, overlayfs's
copy-up), so a spliced reply could return stale data where the copy
path returns current data. Nor on an fd opened O_DIRECT, regardless
of filesystem: the kernel serves that splice by direct I/O into
freshly allocated pipe pages (as pread does, so nothing is stale),
which saves no copy over the copy path while adding syscalls, and a
misaligned request's EINVAL would look like a filesystem without
splice_read. A refusal by filesystem type is remembered for dev_ (the
fd's device, 0 if unknown) so later handles on that filesystem skip
the fstatfs. */
static
bool
_splice_coherent(const int fd_,
const dev_t dev_)
{
#if defined(__linux__)
constexpr u32 MAGIC_FUSE = 0x65735546; // fuse, fuseblk, virtiofs
constexpr u32 MAGIC_CIFS = 0xFF534D42;
constexpr u32 MAGIC_SMB2 = 0xFE534D42;
constexpr u32 MAGIC_NFS = 0x00006969;
constexpr u32 MAGIC_CEPH = 0x00C36400;
constexpr u32 MAGIC_AFS = 0x5346414F; // OpenAFS (/afs)
constexpr u32 MAGIC_KAFS = 0x6B414653; // in-kernel AFS client
constexpr u32 MAGIC_OVERLAYFS = 0x794C7630;
constexpr u32 MAGIC_V9FS = 0x01021997;
struct statfs st;
int rv;
if(::fstatfs(fd_,&st) < 0)
return false;
switch((u32)st.f_type)
{
case MAGIC_FUSE:
case MAGIC_CIFS:
case MAGIC_SMB2:
case MAGIC_NFS:
case MAGIC_CEPH:
case MAGIC_AFS:
case MAGIC_KAFS:
case MAGIC_OVERLAYFS:
case MAGIC_V9FS:
if(dev_ != 0)
g_refused_devs.emplace(dev_,true);
return false;
default:
break;
}
rv = fs::getfl(fd_);
if((rv >= 0) && (rv & O_DIRECT))
return false;
return true;
#else
(void)fd_;
(void)dev_;
return false;
#endif
}
/* Whether [offset, offset+size) lies inside [0, bound), computed
without overflow: offset+size itself can exceed INT64_MAX for reads
near the top of a huge sparse file. */
static
inline
bool
_window_inside(const off_t offset_,
const size_t size_,
const s64 bound_)
{
return ((offset_ >= 0) &&
(offset_ < bound_) &&
((u64)size_ <= (u64)(bound_ - offset_)));
}
/* Describe a large read as a region of the branch file so libfuse can
splice it into /dev/fuse without the data crossing userspace.
Returning a positive value declines, and libfuse calls FUSE::read.
The reply header, whose length is the region size, is staged before
any data, so only a read lying entirely within the file's reported
size is offered. A read crossing it declines rather than being
trimmed: some branches report a size that lags the data
(attribute-caching network or FUSE filesystems), and a trimmed reply
would end short where pread returns the rest. This also declines
non-regular files and files reporting size 0 (procfs) or 4096
(sysfs) while returning live content, which a splice past the
reported size would generate twice or block on. A file that shrinks
between the size check and the splice falls back to FUSE::read.
The size check is skipped when the window ends inside
fi->size_floor, a size already observed on this handle; a stale floor
only costs a fallback, whose short FUSE::read lowers it.
fi->mutex is held shared so moveonenospc's dup2 cannot rebind fd
while libfuse stages from it (see FileRegion::lock). It is never
waited for: a move holds it for a whole file copy, so the read
declines instead and the copy path serves it from the old fd. */
int
FUSE::read_buf(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_,
FileRegion *region_,
const size_t size_,
const off_t offset_)
{
FileInfo *fi;
FileInfo::Splice splice;
s64 floor;
dev_t dev = 0;
fi = state.get_fi(ctx_,ffi_->fh);
if(not fi)
return -EBADF;
if(fi->splice_state.load(std::memory_order_relaxed) == FileInfo::Splice::no)
return 1;
std::shared_lock<std::shared_mutex> lock(fi->mutex,std::try_to_lock);
if(not lock.owns_lock())
return 1;
/* Checked after the lock is held so a mover that started waiting
before try_to_lock succeeded is not joined by this splice. */
if(fi->exclusive_waiters.load(std::memory_order_relaxed) > 0)
return 1;
/* Read again under the lock: a move that completed before try_to_lock
rebound fd and reset this to unknown (fd_rebound), and the
once-per-binding check below must see that. */
splice = fi->splice_state.load(std::memory_order_relaxed);
if(splice == FileInfo::Splice::no)
return 1;
floor = fi->size_floor.load(std::memory_order_relaxed);
if(not ::_window_inside(offset_,size_,floor))
{
mode_t mode;
s64 fsize;
if((::_type_and_size(fi->fd,&mode,&fsize,&dev) < 0) ||
!S_ISREG(mode))
return 1;
/* On a filesystem already refused by type, settle this handle
now, even for a read that would decline below anyway (every
read of a file smaller than the request does), so its later
reads skip this statx. */
if((splice == FileInfo::Splice::unknown) && g_refused_devs.contains(dev))
{
fi->splice_state.compare_exchange_strong(splice,
FileInfo::Splice::no,
std::memory_order_relaxed);
return 1;
}
/* Only replace the floor this check started from: a short read
may have lowered it meanwhile, and that is newer than fsize. */
if(fsize > 0)
fi->size_floor.compare_exchange_strong(floor,fsize,
std::memory_order_relaxed);
if(not ::_window_inside(offset_,size_,fsize))
return 1;
}
/* Once per fd binding, for a read that could otherwise splice, and
under the lock so the answer is about the file fd refers to. A
concurrent rejection (read_buf_rejected) wins over this check. */
if(splice == FileInfo::Splice::unknown)
{
/* A filesystem that cannot be identified is not offered either,
and that is settled too: retrying would repeat the fstatfs,
possibly a server round trip, on every later large read. */
const FileInfo::Splice want = (::_splice_coherent(fi->fd,dev) ?
FileInfo::Splice::yes :
FileInfo::Splice::no);
/* On failure splice holds the concurrent decision, which wins. */
if(fi->splice_state.compare_exchange_strong(splice,want,
std::memory_order_relaxed))
splice = want;
if(splice != FileInfo::Splice::yes)
return 1;
}
/* ioprio is thread-persistent; without SetFrom the branch-file I/O
performed downstream of this handler (the kernel-side splice in
the reply path runs on this same worker thread) would inherit
whatever priority a previous request left behind. */
ioprio::SetFrom iop(ctx_->pid);
region_->fd = fi->fd;
region_->offset = offset_;
region_->size = size_;
region_->lock = std::move(lock);
return 0;
}
/* The branch fd refused splice (EINVAL on its first splice: its
filesystem has no splice_read). That will not change for this fd, so
stop offering it and let every later read go straight to FUSE::read.
O_DIRECT fds never get here: _splice_coherent refuses them. libfuse
calls this while the region's lock is still held, so fd is still the
one that refused. */
void
FUSE::read_buf_rejected(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_)
{
FileInfo *fi;
fi = state.get_fi(ctx_,ffi_->fh);
if(not fi)
return;
fi->splice_state.store(FileInfo::Splice::no,std::memory_order_relaxed);
}
+35
View File
@@ -0,0 +1,35 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#pragma once
#include "fuse_io.hpp"
#include "fuse.h"
namespace FUSE
{
int read_buf(const fuse_req_ctx_t *ctx,
const fuse_file_info_t *ffi,
FileRegion *region,
const size_t size,
const off_t offset);
void read_buf_rejected(const fuse_req_ctx_t *ctx,
const fuse_file_info_t *ffi);
}
+21 -15
View File
@@ -48,30 +48,29 @@ FUSE::ReadDir::ReadDir(const std::string_view s_)
std::string
FUSE::ReadDir::to_string() const
{
std::shared_ptr<std::string> str;
std::shared_lock<std::shared_mutex> lk(_mutex);
str = std::atomic_load(&_str);
return (*str);
return _str;
}
int
FUSE::ReadDir::from_string(const std::string_view str_)
{
std::shared_ptr<std::string> tmp_str;
std::shared_ptr<FUSE::ReadDirBase> tmp_readdir;
tmp_readdir = FUSE::ReadDirFactory::make(str_);
if(!tmp_readdir)
return -EINVAL;
tmp_str = std::make_shared<std::string>(str_);
std::atomic_store(&_str,tmp_str);
{
std::unique_lock<std::shared_mutex> lk(_mutex);
if(_initialized == false)
return 0;
_str = str_;
if(_initialized == false)
return 0;
std::atomic_store(&_impl,tmp_readdir);
_impl = std::move(tmp_readdir);
}
return 0;
}
@@ -114,7 +113,10 @@ FUSE::ReadDir::operator()(const fuse_req_ctx_t *ctx_,
int rv;
std::shared_ptr<FUSE::ReadDirBase> readdir;
readdir = std::atomic_load(&_impl);
{
std::shared_lock<std::shared_mutex> lk(_mutex);
readdir = _impl;
}
if(!readdir)
fatal::abort("readdir impl is null");
@@ -128,10 +130,14 @@ FUSE::ReadDir::operator()(const fuse_req_ctx_t *ctx_,
void
FUSE::ReadDir::initialize()
{
std::shared_ptr<std::string> str;
std::string str;
str = std::atomic_load(&_str);
_initialized = true;
{
std::unique_lock<std::shared_mutex> lk(_mutex);
from_string(*str);
str = _str;
_initialized = true;
}
from_string(str);
}
+4 -3
View File
@@ -22,9 +22,9 @@
#include "tofrom_string.hpp"
#include "fuse_readdir_base.hpp"
#include <atomic>
#include <memory>
#include <mutex>
#include <shared_mutex>
namespace FUSE
@@ -59,8 +59,9 @@ namespace FUSE
void initialize();
private:
std::atomic<bool> _initialized = false;
std::shared_ptr<std::string> _str;
bool _initialized = false;
mutable std::shared_mutex _mutex;
std::string _str;
std::shared_ptr<FUSE::ReadDirBase> _impl;
};
}
+6
View File
@@ -43,6 +43,12 @@ FUSE::ReadDirCOR::~ReadDirCOR()
#include "fuse_readdir_cor_readdir.icpp"
#endif
// Uses a per-call mutex. Lifetime is one readdir operation. It
// serializes the dirent-merge step (dedup'ing of names + append to
// dirents list) across at-most-branches-worth of parallel worker
// tasks. Per-branch opendir + getdents64 run fully outside the
// critical section, so the lock holds only long enough to merge each
// batch.
static
inline
int
+12 -2
View File
@@ -81,11 +81,21 @@ FUSE::release(cu64 nodeid_,
FileInfo *canonical_fi = nullptr;
FileInfo *fh_fi_to_free = nullptr;
// memory_order_acq_rel on the decrement gives us the canonical
// refcount-publication pattern:
// - release: prior writes to the OpenFile by this thread are
// visible to whichever thread observes the final 0.
// - acquire: the thread that drops the last ref acquires every
// prior ref holder's writes before tearing down the entry.
// The concurrent_flat_map bucket lock already serializes concurrent
// visitors of this key so the acq_rel is good for any observer that
// might examine ref_count outside a visit().
of.erase_if(nodeid_,
[&](auto &v_)
{
v_.second.ref_count--;
if(v_.second.ref_count > 0)
const int prev =
v_.second.ref_count.fetch_sub(1,std::memory_order_acq_rel);
if(prev > 1)
{
if(fi_ != v_.second.fi)
fh_fi_to_free = fi_;
+107 -79
View File
@@ -28,11 +28,16 @@
#include "fs_pwriten.hpp"
#include "ioprio.hpp"
#include "state.hpp"
#include "syslog.hpp"
#include "scope_guard/scope_guard.hpp"
#include "fuse.h"
#include <cstring>
#include <mutex>
#include <shared_mutex>
static
bool
@@ -42,6 +47,41 @@ _out_of_space(const ssize_t err_)
(err_ == -EDQUOT));
}
// moveonenospc: move the file to another branch and rebind fi_->fd to
// the copy. Called with fi_->mutex held exclusively. 0 or -errno.
static
int
_move_file(FileInfo *fi_)
{
int err;
int fd;
if(cfg.moveonenospc.enabled == false)
return -ENOSPC;
fd = fs::movefile_and_open_as_root(cfg.moveonenospc.policy,
cfg.branches,
fi_->branch.path,
fi_->fusepath,
fi_->fd);
if(fd < 0)
return fd;
err = fs::dup2(fd,fi_->fd);
{
const int close_err = fs::close(fd);
if(close_err < 0)
SysLog::warning("moveonenospc: closing the copy's fd after the move"
" failed: {}",strerror(-close_err));
}
if(err < 0)
return err;
fi_->fd_rebound();
return 0;
}
static
int
_move_and_pwrite(const char *buf_,
@@ -50,23 +90,7 @@ _move_and_pwrite(const char *buf_,
FileInfo *fi_,
int err_)
{
int err;
ssize_t rv;
if(cfg.moveonenospc.enabled == false)
return err_;
rv = fs::movefile_and_open_as_root(cfg.moveonenospc.policy,
cfg.branches,
fi_->branch.path,
fi_->fusepath,
fi_->fd);
if(rv < 0)
return err_;
err = fs::dup2(rv,fi_->fd);
fs::close(rv);
if(err < 0)
if(::_move_file(fi_) < 0)
return err_;
return fs::pwrite(fi_->fd,buf_,count_,offset_);
@@ -84,20 +108,7 @@ _move_and_pwriten(char const *buf_,
int err;
ssize_t rv;
if(cfg.moveonenospc.enabled == false)
return err_;
rv = fs::movefile_and_open_as_root(cfg.moveonenospc.policy,
cfg.branches,
fi_->branch.path,
fi_->fusepath,
fi_->fd);
if(rv < 0)
return err_;
err = fs::dup2(rv,fi_->fd);
fs::close(rv);
if(err < 0)
if(::_move_file(fi_) < 0)
return err_;
rv = fs::pwriten(fi_->fd,
@@ -117,47 +128,11 @@ _move_and_pwriten(char const *buf_,
// N bytes written (short writes included)
// -errno on error
// See libfuse/include/fuse.h for more details
static
int
_write_direct_io(const char *buf_,
const size_t count_,
const off_t offset_,
FileInfo *fi_)
{
ssize_t rv;
rv = fs::pwrite(fi_->fd,buf_,count_,offset_);
if(::_out_of_space(rv))
rv = ::_move_and_pwrite(buf_,count_,offset_,fi_,rv);
return rv;
}
// When not in direct_io mode write's return value is more complex.
// 0 or less than `count` on EOF
// `count` on non-errors
// -errno on error
// See libfuse/include/fuse.h for more details
static
int
_write_cached(const char *buf_,
const size_t count_,
const off_t offset_,
FileInfo *fi_)
{
int err;
ssize_t rv;
rv = fs::pwriten(fi_->fd,buf_,count_,offset_,&err);
if(err == 0)
return rv;
if(err && !::_out_of_space(err))
return err;
rv = ::_move_and_pwriten(buf_,count_,offset_,fi_,err,rv);
return rv;
}
static
int
@@ -175,19 +150,60 @@ _write(const fuse_req_ctx_t *ctx_,
// Concurrent writes can only happen if:
// 1) writeback-cache is enabled and using page caching
// 2) parallel_direct_writes is enabled and file has `direct_io=true`
// Will look into selectively locking in the future
// A reader/writer lock would probably be the best option given
// the expense of the write itself in comparison. Alternatively,
// could change the move file behavior to use a known target file
// and have threads use O_EXCL and back off and wait for the
// transfer to complete before retrying.
mutex_lockguard(fi->mutex);
// 2) parallel_direct_writes is enabled and file has
// `direct_io=true`
if(fi->direct_io)
return ::_write_direct_io(buf_,count_,offset_,fi);
ssize_t rv;
return ::_write_cached(buf_,count_,offset_,fi);
// A writeback (writepage) request must be written completely or
// fail: the kernel ignores a short count for it (see FUSE::write).
// So even on a direct_io handle it takes the looping path below,
// which also reaches moveonenospc when the branch fills mid-write.
if(fi->direct_io && !ffi_->writepage)
{
{
std::shared_lock<std::shared_mutex> slk(fi->mutex);
rv = fs::pwrite(fi->fd,buf_,count_,offset_);
}
if(not ::_out_of_space(rv))
return rv;
std::unique_lock<std::shared_mutex> ulk = fi->lock_exclusive();
// Re-check under exclusive lock: another writer may have
// already moved the file and retired. If so the pwrite should
// now succeed without our own move call.
rv = fs::pwrite(fi->fd,buf_,count_,offset_);
if(::_out_of_space(rv))
rv = ::_move_and_pwrite(buf_,count_,offset_,fi,rv);
return rv;
}
else
{
int err;
ssize_t written;
{
std::shared_lock<std::shared_mutex> slk(fi->mutex);
written = fs::pwriten(fi->fd,buf_,count_,offset_,&err);
}
if(err == 0)
return written;
if(err && not ::_out_of_space(err))
return err;
std::unique_lock<std::shared_mutex> ulk = fi->lock_exclusive();
// Re-check under exclusive lock as another move may have
// already run.
written = fs::pwriten(fi->fd,buf_,count_,offset_,&err);
if(err == 0)
return written;
if(err && not ::_out_of_space(err))
return err;
return ::_move_and_pwriten(buf_,count_,offset_,fi,err,written);
}
}
int
@@ -198,8 +214,20 @@ FUSE::write(const fuse_req_ctx_t *ctx_,
off_t offset_)
{
ioprio::SetFrom iop(ctx_->pid);
int rv;
return ::_write(ctx_,ffi_,buf_,count_,offset_);
rv = ::_write(ctx_,ffi_,buf_,count_,offset_);
/* Writeback completion (FUSE_WRITE_CACHE) ignores a short reply's byte
count: Linux fuse_writepage_end consults only the error, so a short
nonnegative result would mark the unwritten tail clean and lose it.
_write already loops for these requests; a short count here means
a pwrite made no progress without an error, so report a failure.
Ordinary writes report their prefix. */
if(ffi_->writepage && (rv >= 0) && ((size_t)rv < count_))
return -EIO;
return rv;
}
int
+8 -12
View File
@@ -32,8 +32,12 @@ enum
IOPRIO_WHO_PROCESS = 1
};
namespace ioprio
{
std::atomic<bool> _enabled{false};
}
thread_local int ioprio::SetFrom::thread_prio = -1;
bool _enabled = false;
int
ioprio::get(const int who_)
@@ -69,22 +73,14 @@ ioprio::set(const int who_,
void
ioprio::enable(const bool enable_)
{
_enabled = enable_;
_enabled.store(enable_,std::memory_order_relaxed);
}
bool
ioprio::enabled()
{
return _enabled;
}
ioprio::SetFrom::SetFrom(const pid_t pid_)
void
ioprio::SetFrom::_slow_apply(const pid_t pid_)
{
int client_prio;
if(!_enabled)
return;
client_prio = ioprio::get(pid_);
if(client_prio < 0)
return;
+25 -2
View File
@@ -18,13 +18,27 @@
#pragma once
#include <atomic>
#include <sys/types.h>
namespace ioprio
{
// Exposed so the disabled-path check in SetFrom can inline into
// every fuse_read/fuse_write as an atomic load + branch, with no
// function overhead.
extern std::atomic<bool> _enabled;
[[gnu::always_inline]]
inline
bool
enabled()
{
return _enabled.load(std::memory_order_relaxed);
}
void enable(const bool);
bool enabled();
int get(const int who);
int set(const int who, const int ioprio);
@@ -33,6 +47,15 @@ namespace ioprio
{
static thread_local int thread_prio;
SetFrom(const pid_t pid);
[[gnu::always_inline]]
inline
SetFrom(const pid_t pid_)
{
if(ioprio::enabled())
_slow_apply(pid_);
}
private:
void _slow_apply(const pid_t);
};
};
+6 -1
View File
@@ -63,6 +63,7 @@
#include "fuse_opendir.hpp"
#include "fuse_poll.hpp"
#include "fuse_read.hpp"
#include "fuse_read_buf.hpp"
#include "fuse_readdir.hpp"
#include "fuse_readdir_plus.hpp"
#include "fuse_readlink.hpp"
@@ -126,6 +127,10 @@ _get_fuse_operations(struct fuse_operations &ops_,
ops_.opendir = FUSE::opendir;
ops_.poll = FUSE::poll;
ops_.read = (nullrw_ ? FUSE::read_null : FUSE::read);
/* Large reads are spliced from the branch file into /dev/fuse when
possible; nullrw has no branch data to splice. */
ops_.read_buf = (nullrw_ ? nullptr : FUSE::read_buf);
ops_.read_buf_rejected = (nullrw_ ? nullptr : FUSE::read_buf_rejected);
ops_.readdir = FUSE::readdir;
ops_.readdir_plus = FUSE::readdir_plus;
ops_.readlink = FUSE::readlink;
@@ -301,7 +306,7 @@ _warn_if_not_root()
"mergerfs is not running as root"
" and may not work correctly";
fmt::println(stderr,"* WARNING: {}",s);
SysLog::warning(s);
SysLog::warning("{}",s);
}
int
+5 -1
View File
@@ -103,8 +103,12 @@ public:
{
}
// Move constructor runs during initial insertion into the
// concurrent_flat_map before the new entry is reachable by the
// other threads. There is no concurrent observer of the moved
// value at this point so relaxed load is sufficient.
OpenFile(OpenFile &&o_) noexcept
: ref_count(o_.ref_count.load()),
: ref_count(o_.ref_count.load(std::memory_order_relaxed)),
backing_id(o_.backing_id),
fi(o_.fi)
{
+712
View File
@@ -1,6 +1,13 @@
#include "acutest/acutest.h"
#include "config.hpp"
#include "fileinfo.hpp"
#include "fuse_cfg.hpp"
#include "fuse_ftruncate.hpp"
#include "fuse_lowlevel.h"
#include "fuse_read.hpp"
#include "fuse_read_buf.hpp"
#include "fuse_write.hpp"
#include "fs_copyfile.hpp"
#include "fs_inode.hpp"
#include "from_string.hpp"
@@ -27,6 +34,18 @@
#include <errno.h>
#include <sys/stat.h>
#include <unistd.h>
#include <signal.h>
#include <sys/resource.h>
#include <sys/wait.h>
#if defined(__linux__)
#include <linux/filter.h>
#include <linux/seccomp.h>
#include <stddef.h>
#include <sys/eventfd.h>
#include <sys/prctl.h>
#include <sys/syscall.h>
#include <sys/vfs.h>
#endif
template<typename Predicate>
bool
@@ -50,6 +69,685 @@ test_nop()
TEST_CHECK(true);
}
#if defined(__linux__)
/* Drive fuse_reply_data_splice_fd with a regular file standing in for
/dev/fuse and return every byte the "kernel" received. Unlike a pipe,
the file takes the whole message whatever pipe-max-size or the
per-user pipe quota allow, so a short result is the reply path's. */
static
std::string
_splice_reply_bytes(const int fd_,
const off_t offset_,
const size_t size_,
FuseReplyResult *result_,
const uint64_t unique_ = 7)
{
FileRegion region{fd_,offset_,size_,{}};
*result_ = FuseReplyResult::failed;
FILE *dev = tmpfile();
if(dev == nullptr)
return "tmpfile failed";
fuse_req_t *req = fuse_req_alloc();
req->fd = fileno(dev);
req->ctx.unique = unique_;
*result_ = fuse_reply_data_splice_fd(req,region);
fuse_req_free(req);
std::string got;
char buf[65536];
ssize_t n;
while((n = pread(fileno(dev),buf,sizeof(buf),(off_t)got.size())) > 0)
got.append(buf,n);
fclose(dev);
return got;
}
/* Whether this process may grow a pipe to bytes_, as the reply path
must for the tests below: unprivileged, pipe-max-size or the per-user
pipe quota can refuse it, and the reply then correctly falls back. */
static
bool
_pipe_can_grow(const size_t bytes_)
{
int fds[2];
if(pipe2(fds,O_CLOEXEC) != 0)
return false;
const int rv = fcntl(fds[0],F_SETPIPE_SZ,(int)bytes_);
close(fds[0]);
close(fds[1]);
return (rv >= (int)bytes_);
}
/* A tmpfile() holding size_ bytes of data_, or nullptr. */
static
FILE*
_tmpfile_with(const void *data_,
const size_t size_)
{
FILE *file = tmpfile();
if((file != nullptr) &&
(pwrite(fileno(file),data_,size_,0) != (ssize_t)size_))
{
fclose(file);
return nullptr;
}
return file;
}
/* Whether got_ is exactly one successful FUSE message for unique_
carrying the len_ bytes at expected_. */
static
bool
_is_reply(const std::string &got_,
const uint64_t unique_,
const char *expected_,
const size_t len_)
{
fuse_out_header hdr;
if(got_.size() != sizeof(hdr) + len_)
return false;
memcpy(&hdr,got_.data(),sizeof(hdr));
return ((hdr.len == sizeof(hdr) + len_) &&
(hdr.unique == unique_) &&
(hdr.error == 0) &&
(memcmp(got_.data() + sizeof(hdr),expected_,len_) == 0));
}
/* An open handle on fd_ as FUSE::read_buf and friends see it. */
struct TestHandle
{
Branch branch;
FileInfo fi;
fuse_req_ctx_t ctx = {};
fuse_file_info_t ffi = {};
TestHandle(const int fd_,
const char *fusepath_)
: fi(fd_,branch,fs::path(fusepath_),false)
{
ffi.fh = fi.to_fh();
}
};
/* A spliced reply must be exactly one FUSE message: a header whose len
covers the payload, the right unique, zero error, then the file bytes
of the requested window, including an unaligned start. */
void
test_reply_splice_sends_exact_message()
{
if(!::_pipe_can_grow(262144))
{
TEST_SKIP("cannot grow a pipe to 256KiB here");
return;
}
static char data[300000];
for(size_t i = 0; i < sizeof(data); i++)
data[i] = (char)(i * 31 + 7);
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
const off_t off = 4097;
const size_t len = 131072;
FuseReplyResult r;
const std::string got = _splice_reply_bytes(fileno(file),off,len,&r,42);
TEST_CHECK(r == FuseReplyResult::sent);
TEST_CHECK(::_is_reply(got,42,data + off,len));
fclose(file);
}
/* A region that runs past end-of-file must not send a short message:
nothing may reach the kernel and the caller gets fallback, so it can
reply through read(). */
void
test_reply_splice_eof_falls_back_without_sending()
{
if(!::_pipe_can_grow(262144))
{
TEST_SKIP("cannot grow a pipe to 256KiB here");
return;
}
static char data[70000];
for(size_t i = 0; i < sizeof(data); i++)
data[i] = (char)(i * 13 + 5);
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
FuseReplyResult r;
const std::string got = _splice_reply_bytes(fileno(file),0,131072,&r);
TEST_CHECK(r == FuseReplyResult::fallback);
TEST_CHECK(got.empty());
/* The abandoned staging must leave this thread's reply pipe empty:
the next reply is exactly its own message, nothing left over. */
const std::string got2 = _splice_reply_bytes(fileno(file),4096,65536,&r,9);
TEST_CHECK(r == FuseReplyResult::sent);
TEST_CHECK(::_is_reply(got2,9,data + 4096,65536));
fclose(file);
}
/* A reply below the splice threshold is cheaper through read + writev,
so the splice path declines it before staging anything. */
void
test_reply_splice_small_region_falls_back()
{
char data[8192];
memset(data,0x44,sizeof(data));
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
FuseReplyResult r;
std::string got = _splice_reply_bytes(fileno(file),0,8192,&r);
TEST_CHECK(r == FuseReplyResult::fallback);
TEST_CHECK(got.empty());
got = _splice_reply_bytes(fileno(file),8192,0,&r);
TEST_CHECK(r == FuseReplyResult::fallback);
TEST_CHECK(got.empty());
fclose(file);
}
/* A descriptor splice rejects with EINVAL is reported as unsupported,
again with nothing sent. An eventfd has no pread, so splicing from
it at an offset is EINVAL on every kernel (a directory is EINVAL only
since Linux 5.10; older kernels fail it with EISDIR). */
void
test_reply_splice_unspliceable_fd_is_unsupported()
{
if(!::_pipe_can_grow(131072))
{
TEST_SKIP("cannot grow a pipe to 128KiB here");
return;
}
const int efd = eventfd(0,EFD_CLOEXEC);
TEST_ASSERT(efd >= 0);
FuseReplyResult r;
const std::string got = _splice_reply_bytes(efd,0,65536,&r);
TEST_CHECK(r == FuseReplyResult::unsupported);
TEST_CHECK(got.empty());
close(efd);
}
/* The region's lock is released as FileRegion::lock documents: once the
data is staged for a sent reply, but kept by a fallback during
staging (`tail` runs past end-of-file). */
void
test_reply_splice_releases_lock_before_submit()
{
if(!::_pipe_can_grow(262144))
{
TEST_SKIP("cannot grow a pipe to 256KiB here");
return;
}
static char data[200000];
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
FILE *dev = tmpfile();
TEST_ASSERT(dev != nullptr);
std::shared_mutex mutex;
fuse_req_t *req = fuse_req_alloc();
req->fd = fileno(dev);
req->ctx.unique = 11;
FileRegion sent{fileno(file),0,131072,std::shared_lock<std::shared_mutex>(mutex)};
TEST_CHECK(fuse_reply_data_splice_fd(req,sent) == FuseReplyResult::sent);
TEST_CHECK(!sent.lock.owns_lock());
TEST_CHECK(mutex.try_lock());
mutex.unlock();
FileRegion tail{fileno(file),131072,131072,std::shared_lock<std::shared_mutex>(mutex)};
TEST_CHECK(fuse_reply_data_splice_fd(req,tail) == FuseReplyResult::fallback);
TEST_CHECK(tail.lock.owns_lock());
fuse_req_free(req);
fclose(dev);
fclose(file);
}
/* A debug log write that throws once the reply is fully staged (fmt on
a full or broken log stream) must not leave that message in this
thread's reply pipe: the worker survives the exception, and its next
reply would otherwise be sent behind, or instead of, the stale one. */
void
test_reply_splice_debug_throw_leaves_pipe_empty()
{
if(!::_pipe_can_grow(262144))
{
TEST_SKIP("cannot grow a pipe to 256KiB here");
return;
}
FILE *full = fopen("/dev/full","w");
if(full == nullptr)
{
TEST_SKIP("no /dev/full here");
return;
}
setvbuf(full,nullptr,_IONBF,0);
static char data[200000];
for(size_t i = 0; i < sizeof(data); i++)
data[i] = (char)(i * 7 + 3);
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
FILE *dev = tmpfile();
TEST_ASSERT(dev != nullptr);
const std::shared_ptr<FILE> log_file = fuse_cfg.log_file();
const bool debug = fuse_cfg.debug;
fuse_cfg.log_file(std::shared_ptr<FILE>(full,fclose));
fuse_cfg.debug = true;
fuse_req_t *req = fuse_req_alloc();
req->fd = fileno(dev);
req->ctx.unique = 5;
FileRegion region{fileno(file),0,131072,{}};
bool threw = false;
try
{
fuse_reply_data_splice_fd(req,region);
}
catch(...)
{
threw = true;
}
fuse_req_free(req);
fuse_cfg.debug = debug;
fuse_cfg.log_file(log_file);
struct stat st;
TEST_CHECK(threw);
TEST_CHECK((fstat(fileno(dev),&st) == 0) && (st.st_size == 0));
FuseReplyResult r;
const std::string got = _splice_reply_bytes(fileno(file),4096,65536,&r,9);
TEST_CHECK(r == FuseReplyResult::sent);
TEST_CHECK(::_is_reply(got,9,data + 4096,65536));
fclose(dev);
fclose(file);
}
/* Whether fd_ is on a local filesystem that read_buf must always offer
for splicing (ext2/3/4, XFS, tmpfs, btrfs, f2fs). */
static
bool
_splice_local_fs(const int fd_)
{
struct statfs st;
if(::fstatfs(fd_,&st) != 0)
return false;
switch((unsigned long)st.f_type)
{
case 0xEF53: // ext2/3/4
case 0x58465342: // xfs
case 0x01021994: // tmpfs
case 0x9123683E: // btrfs
case 0xF2F52010: // f2fs
return true;
default:
return false;
}
}
/* Whether read_buf refuses the filesystem tmpfile() uses outright
(overlayfs, NFS, FUSE, ...: see _splice_coherent). tmpfile() always
uses /tmp, which is such a filesystem inside most containers, so the
read_buf tests skip there instead of failing. Never on a local
filesystem read_buf must offer: a refusal there is the regression
the tests exist to catch, so they run (and fail) instead of every
one of them skipping. Probes its own file through a throwaway
handle, before the caller performs any check (acutest cannot skip
after one). */
static
bool
_read_buf_refuses_tmp()
{
static char zeros[65536];
FILE *file = ::_tmpfile_with(zeros,sizeof(zeros));
if(file == nullptr)
return false;
TestHandle h(fileno(file),"/read-buf-probe");
FileRegion region;
FUSE::read_buf(&h.ctx,&h.ffi,&region,sizeof(zeros),0);
const bool refused = (h.fi.splice_state.load() == FileInfo::Splice::no);
const bool local = ::_splice_local_fs(fileno(file));
region = {};
fclose(file);
return (refused && not local);
}
/* read_buf hands out the branch fd and the whole request window, and
only for a window lying entirely within the file's reported size:
the header carrying the region size is staged before any data, and
a reported size may lag the data, so a window crossing it (a tail,
an EOF probe, any read of a size-0 file) declines instead of being
trimmed. A rejected handle declines too; a bad handle fails; a window
below the splice threshold is never asked for. The returned region
holds the handle's lock shared so the fd cannot be rebound
mid-splice. */
void
test_read_buf_region_contract()
{
if(::_read_buf_refuses_tmp())
{
TEST_SKIP("read_buf refuses this /tmp filesystem");
return;
}
static char data[200000];
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
TestHandle h(fileno(file),"/read-buf");
FileInfo &fi = h.fi;
const fuse_req_ctx_t &ctx = h.ctx;
const fuse_file_info_t &ffi = h.ffi;
FileRegion region;
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,65536,1234) == 0);
TEST_CHECK(region.fd == fileno(file));
TEST_CHECK(region.offset == 1234 && region.size == 65536);
TEST_CHECK(region.lock.owns_lock() && region.lock.mutex() == &fi.mutex);
TEST_CHECK(!fi.mutex.try_lock());
region = {};
TEST_CHECK(fi.mutex.try_lock());
fi.mutex.unlock();
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,68928) == 0);
TEST_CHECK(region.offset == 68928 && region.size == 131072);
region = {};
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,68929) > 0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,150000) > 0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,200000) > 0);
TEST_CHECK(!region.lock.owns_lock());
TEST_CHECK(fi.mutex.try_lock());
fi.mutex.unlock();
FILE *empty = tmpfile();
TEST_ASSERT(empty != nullptr);
TestHandle eh(fileno(empty),"/read-buf-empty");
TEST_CHECK(FUSE::read_buf(&eh.ctx,&eh.ffi,&region,131072,0) > 0);
fclose(empty);
region = {};
FUSE::read_buf_rejected(&ctx,&ffi);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,0) > 0);
fuse_file_info_t bad = {};
TEST_CHECK(FUSE::read_buf(&ctx,&bad,&region,65536,0) == -EBADF);
fclose(file);
}
/* read_buf never waits for the handle's lock: while moveonenospc holds
it, or waits for it, a read declines to the copy path (which reads
the old fd, as it always has) rather than pinning a worker thread for
the whole move, or holding the mover off. */
void
test_read_buf_never_waits_for_mover()
{
if(::_read_buf_refuses_tmp())
{
TEST_SKIP("read_buf refuses this /tmp filesystem");
return;
}
static char data[200000];
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
TestHandle h(fileno(file),"/read-buf-mover");
FileInfo &fi = h.fi;
const fuse_req_ctx_t &ctx = h.ctx;
const fuse_file_info_t &ffi = h.ffi;
FileRegion region;
fi.mutex.lock();
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,65536,0) > 0);
fi.mutex.unlock();
fi.exclusive_waiters.store(1);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,65536,0) > 0);
fi.exclusive_waiters.store(0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,65536,0) == 0);
TEST_CHECK(region.lock.owns_lock());
region = {};
fclose(file);
}
/* An O_DIRECT branch fd is never offered, whatever its filesystem: the
kernel would serve that splice by direct I/O into fresh pages, saving
no copy over the copy path, and a misaligned request's EINVAL would
look like a filesystem without splice_read. */
void
test_read_buf_declines_odirect()
{
if(::_read_buf_refuses_tmp())
{
TEST_SKIP("read_buf refuses this /tmp filesystem");
return;
}
FILE *file = tmpfile();
char path[64];
int dfd = -1;
if(file != nullptr)
{
snprintf(path,sizeof(path),"/proc/self/fd/%d",fileno(file));
dfd = open(path,O_RDONLY|O_DIRECT);
}
if(dfd < 0)
{
if(file != nullptr)
fclose(file);
TEST_SKIP("no O_DIRECT tmpfile here");
return;
}
static char data[200000];
TEST_ASSERT(pwrite(fileno(file),data,sizeof(data),0) == sizeof(data));
TestHandle h(dfd,"/read-buf-odirect");
FileRegion region;
TEST_CHECK(FUSE::read_buf(&h.ctx,&h.ffi,&region,65536,0) > 0);
TEST_CHECK(h.fi.splice_state.load() == FileInfo::Splice::no);
close(dfd);
fclose(file);
}
/* The size floor lets read_buf skip its size check. A shrink outside
this handle leaves it stale, so read_buf still offers the whole
window (the reply then hits the end and falls back, see
reply_splice_eof_falls_back_without_sending); a short FUSE::read
lowers it to where that read found the end, and an ftruncate
through the handle resets it. */
void
test_read_buf_size_floor_follows_shrink()
{
if(::_read_buf_refuses_tmp())
{
TEST_SKIP("read_buf refuses this /tmp filesystem");
return;
}
static char data[400000];
FILE *file = ::_tmpfile_with(data,sizeof(data));
TEST_ASSERT(file != nullptr);
TestHandle h(fileno(file),"/read-buf-floor");
FileInfo &fi = h.fi;
const fuse_req_ctx_t &ctx = h.ctx;
const fuse_file_info_t &ffi = h.ffi;
FileRegion region;
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,0) == 0);
region = {};
TEST_ASSERT(ftruncate(fileno(file),100000) == 0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,131072) == 0);
TEST_CHECK(region.size == 131072);
region = {};
static char buf[131072];
TEST_CHECK(FUSE::read(&ctx,&ffi,buf,sizeof(buf),0) == 100000);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,0) > 0);
TEST_ASSERT(pwrite(fileno(file),data,sizeof(data),0) == sizeof(data));
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,131072) == 0);
TEST_CHECK(region.size == 131072);
region = {};
TEST_CHECK(FUSE::ftruncate(&ctx,fi.to_fh(),200000) == 0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,131072) > 0);
TEST_CHECK(FUSE::read_buf(&ctx,&ffi,&region,131072,65536) == 0);
TEST_CHECK(region.size == 131072);
fclose(file);
}
/* A process whose seccomp policy refuses splice (answering EPERM, as
libseccomp, Docker and systemd policies commonly do, rather than
ENOSYS) gets that error from every branch splice. It must turn the
reply splice path off, as a kernel without splice does, instead of
paying for a staging attempt on every large read. Runs in a child:
the filter and the off switch are both process-wide. */
void
test_reply_splice_seccomp_refusal_disables()
{
/* Where the reply pipe cannot grow, the child's staging fails with
EPERM before the filter is involved; that is recorded as a
pipe-max-size bound, which also turns the splice path off, so the
test would pass without testing the seccomp refusal at all. */
if(!::_pipe_can_grow(262144))
{
TEST_SKIP("cannot grow a pipe to 256KiB here");
return;
}
pid_t child = fork();
if(child == 0)
{
alarm(5);
FILE *file = tmpfile();
FILE *dev = tmpfile();
if((file == nullptr) || (dev == nullptr))
_exit(77);
static char data[200000];
if(pwrite(fileno(file),data,sizeof(data),0) != (ssize_t)sizeof(data))
_exit(77);
if(!fuse_reply_splice_fits(0,131072))
_exit(77);
struct sock_filter filter[] =
{
BPF_STMT(BPF_LD|BPF_W|BPF_ABS,offsetof(struct seccomp_data,nr)),
BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K,__NR_splice,0,1),
BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ERRNO|(EPERM & SECCOMP_RET_DATA)),
BPF_STMT(BPF_RET|BPF_K,SECCOMP_RET_ALLOW),
};
struct sock_fprog prog = {(unsigned short)(sizeof(filter) / sizeof(filter[0])),filter};
if((prctl(PR_SET_NO_NEW_PRIVS,1,0,0,0) != 0) ||
(syscall(SYS_seccomp,SECCOMP_SET_MODE_FILTER,0,&prog) != 0))
_exit(77);
fuse_req_t *req = fuse_req_alloc();
req->fd = fileno(dev);
req->ctx.unique = 3;
FileRegion region{fileno(file),0,131072,{}};
const FuseReplyResult r = fuse_reply_data_splice_fd(req,region);
fuse_req_free(req);
struct stat st;
const bool correct = ((r == FuseReplyResult::fallback) &&
(fstat(fileno(dev),&st) == 0) &&
(st.st_size == 0) &&
!fuse_reply_splice_fits(0,131072));
_exit(correct ? 0 : 1);
}
/* No check before this point: acutest cannot skip after one. */
int status = 0;
pid_t waited = -1;
if(child > 0)
do { waited = waitpid(child,&status,0); } while(waited < 0 && errno == EINTR);
if((child > 0) && (waited == child) && WIFEXITED(status) && (WEXITSTATUS(status) == 77))
{
TEST_SKIP("cannot install a seccomp filter or splice here");
return;
}
TEST_CHECK((child > 0) && (waited == child) &&
WIFEXITED(status) && (WEXITSTATUS(status) == 0));
}
/* A writeback request must never end in a short success: Linux's
fuse_writepage_end consults only the error, so a short success would
mark the unwritten tail clean and lose it. RLIMIT_FSIZE=4096 makes an
8KiB direct_io write complete 4096 bytes with no error. An ordinary
request reports that prefix. A writeback request keeps writing, as on
a cached handle (which is also how moveonenospc gets to run when a
branch fills mid-write), and reports the error the remainder gets
(EFBIG). The file holds exactly the 4096-byte prefix either way. */
void
test_write_short_success_writeback()
{
for(int writepage = 0; writepage <= 1; ++writepage)
{
pid_t child = fork();
TEST_ASSERT(child >= 0);
if(child == 0)
{
alarm(5);
signal(SIGXFSZ,SIG_IGN);
struct rlimit limit = {4096,4096};
if(setrlimit(RLIMIT_FSIZE,&limit) != 0)
_exit(1);
FILE *file = tmpfile();
if(file == nullptr)
_exit(2);
char payload[8192];
memset(payload,0x5a,sizeof(payload));
Branch branch;
FileInfo fi(fileno(file),branch,fs::path("/write-short-test"),true);
fuse_req_ctx_t ctx = {};
fuse_file_info_t ffi = {};
ffi.fh = fi.to_fh();
ffi.writepage = writepage;
const int rv = FUSE::write(&ctx,&ffi,payload,sizeof(payload),0);
struct stat st;
char stored[4096];
bool correct = (rv == (writepage ? -EFBIG : 4096)) &&
fstat(fileno(file),&st) == 0 && st.st_size == 4096 &&
pread(fileno(file),stored,sizeof(stored),0) == sizeof(stored) &&
memcmp(stored,payload,sizeof(stored)) == 0;
fclose(file);
_exit(correct ? 0 : 3);
}
int status;
pid_t waited;
do { waited = waitpid(child,&status,0); } while(waited < 0 && errno == EINTR);
TEST_CHECK(waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0);
}
}
#endif
void
test_config_bool()
{
@@ -3650,6 +4348,20 @@ test_rapidhash_withSeed_preserves_default_output()
TEST_LIST =
{
{"nop",test_nop},
#if defined(__linux__)
{"reply_splice_sends_exact_message",test_reply_splice_sends_exact_message},
{"reply_splice_eof_falls_back_without_sending",test_reply_splice_eof_falls_back_without_sending},
{"reply_splice_small_region_falls_back",test_reply_splice_small_region_falls_back},
{"reply_splice_unspliceable_fd_is_unsupported",test_reply_splice_unspliceable_fd_is_unsupported},
{"reply_splice_releases_lock_before_submit",test_reply_splice_releases_lock_before_submit},
{"reply_splice_debug_throw_leaves_pipe_empty",test_reply_splice_debug_throw_leaves_pipe_empty},
{"read_buf_region_contract",test_read_buf_region_contract},
{"read_buf_never_waits_for_mover",test_read_buf_never_waits_for_mover},
{"read_buf_declines_odirect",test_read_buf_declines_odirect},
{"read_buf_size_floor_follows_shrink",test_read_buf_size_floor_follows_shrink},
{"reply_splice_seccomp_refusal_disables",test_reply_splice_seccomp_refusal_disables},
{"write_short_success_writeback",test_write_short_success_writeback},
#endif
{"config_bool",test_config_bool},
{"config_uint64",test_config_uint64},
{"config_int",test_config_int},
+23 -15
View File
@@ -32,9 +32,6 @@ UTILS :=
INSTALLUTILS :=
endif
ifeq ($(NDEBUG),1)
OPT_FLAGS := -O2 -DNDEBUG
else
ifdef SANITIZE
ifeq ($(SANITIZE),1)
override SANITIZE := -fsanitize=address,undefined,leak
@@ -42,15 +39,22 @@ else ifeq ($(filter -fsanitize=%,$(SANITIZE)),)
override SANITIZE := -fsanitize=$(SANITIZE)
endif
endif
OPT_FLAGS := -O0 \
-g \
-fno-omit-frame-pointer \
$(SANITIZE) \
-fstack-protector-strong \
-Wextra \
-Werror \
-Wno-unused-parameter \
-DDEBUG
COMMON_OPT_FLAGS += \
-fstack-protector-strong \
-Wextra \
-Wno-unused-parameter
DEBUG_OPT_FLAGS += \
-O0 \
-g3 \
-fno-omit-frame-pointer \
$(SANITIZE)
ifdef RELEASE
OPT_FLAGS := -O3 $(COMMON_OPT_FLAGS)
else
OPT_FLAGS ?= $(DEBUG_OPT_FLAGS) $(COMMON_OPT_FLAGS)
endif
ifeq ($(LTO),1)
@@ -110,7 +114,7 @@ CXXFLAGS ?= \
-Wall \
-pipe
override CXXFLAGS += \
-std=c++17 \
-std=c++20 \
-MMD \
-MP
FUSERMOUNT_DIR := $(BINDIR)
@@ -151,10 +155,10 @@ $(BUILDDIR)/mount.mergerfs: $(BUILDDIR)/libfuse.a util/mount.mergerfs.cpp
mount.mergerfs: $(BUILDDIR)/mount.mergerfs
$(BUILDDIR)/.objs/%.c.o: lib/%.c | $(BUILDDIR)/stamp
$(CC) $(CFLAGS) $(FUSE_FLAGS) -c $< -o $@
$(CC) $(CFLAGS) $(CPPFLAGS) $(FUSE_FLAGS) -c $< -o $@
$(BUILDDIR)/.objs/%.cpp.o: lib/%.cpp | $(BUILDDIR)/stamp
$(CXX) $(CXXFLAGS) $(FUSE_FLAGS) -c $< -o $@
$(CXX) $(CXXFLAGS) $(CPPFLAGS) $(FUSE_FLAGS) -c $< -o $@
.PHONY: clean
clean:
@@ -176,6 +180,10 @@ install-utils: mergerfs-fusermount mount.mergerfs strip
install: $(INSTALLUTILS)
.PHONY: release
release:
$(MAKE) RELEASE=1 NDEBUG=1 CPPFLAGS="$(CPPFLAGS) -DNDEBUG" all
.PHONY: objects strip utils install install-utils
-include $(DEPS)
+23
View File
@@ -22,6 +22,9 @@
#include <sys/uio.h>
#include <stdint.h>
/* Transport type from fuse_io.hpp (C++); opaque to C consumers. */
struct FileRegion;
EXTERN_C_BEGIN
/* ----------------------------------------------------------- *
@@ -206,6 +209,26 @@ struct fuse_operations
char *buf,
size_t size,
off_t off);
/* Optional. For large reads on Linux: describe the reply as a region
of a branch fd instead of filling a buffer, so libfuse can splice
it into /dev/fuse. Called only for requests whose reply could be
spliced (fuse_reply_splice_fits). Return 0 with region filled to
splice it, a positive value to decline (libfuse then calls read()),
or -errno to fail the request. When returning 0, region->size MUST
equal size: the reply's length is taken from region->size alone,
so a smaller, non-EOF region would be sent as a short reply
indistinguishable from a real short read. To offer less than the
whole request, decline instead and let read() serve it. libfuse
also falls back to read() whenever the splice cannot be staged,
and calls read_buf_rejected when the region's fd refuses splice so
the filesystem can stop offering that open file. */
int (*read_buf)(const fuse_req_ctx_t *,
const fuse_file_info_t *ffi,
struct FileRegion *region,
size_t size,
off_t off);
void (*read_buf_rejected)(const fuse_req_ctx_t *,
const fuse_file_info_t *ffi);
int (*fallocate)(const fuse_req_ctx_t *,
const uint64_t,
int,
+7
View File
@@ -5,6 +5,7 @@
#include <climits>
#include <cstdio>
#include <memory>
#include <shared_mutex>
#include <string>
#define FUSE_CFG_INVALID_ID -1
@@ -33,6 +34,7 @@ struct fuse_cfg_t
std::shared_ptr<std::string> log_filepath() const;
void log_filepath(const std::string &);
private:
mutable std::shared_mutex _log_mutex;
std::shared_ptr<FILE> _log_file = sp<FILE>(stderr,[](FILE*){});
std::shared_ptr<std::string> _log_filepath = sp<std::string>();
@@ -45,6 +47,11 @@ public:
int read_thread_count = 0;
int process_thread_count = -1;
int process_thread_queue_depth = 2;
// After each request a sync FUSE reader polls /dev/fuse for up to this many
// microseconds before blocking in read(). Trades CPU while the mount is
// busy for lower per-request latency on synchronous request streams.
// 0 disables.
u64 read_spin_us = 0;
std::string pin_threads = "false";
u16 request_timeout = 0;
+45
View File
@@ -0,0 +1,45 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#pragma once
#include <cstddef>
#include <shared_mutex>
#include <sys/types.h>
/* A borrowed, descriptor-referenced region of a branch file that a read
reply can splice into /dev/fuse without the data crossing userspace.
The filesystem fills it; libfuse never owns or closes the fd.
The reply splices from fd several times, so the filesystem may park
a shared lock in lock to keep fd bound to the same file until every
read from fd is done (staged into the reply). The reply path
releases it as soon as that staging finishes, whether or not the
reply is then successfully delivered: fd is not read again either
way, and once the READ is answered the kernel may send RELEASE, so
the filesystem may free what the lock lives in. A fallback reached
instead DURING staging (fd itself declined or ran out of data)
leaves the lock held; it is released when the region goes out of
scope, before any fallback read. */
struct FileRegion
{
int fd = -1;
off_t offset = 0;
size_t size = 0;
std::shared_lock<std::shared_mutex> lock;
};
+38 -1
View File
@@ -5,6 +5,17 @@
#include "fuse_kernel.h"
#include "fuse_req.hpp"
#ifdef __cplusplus
#include "fuse_io.hpp"
/* Outcome of an fd-backed reply. fallback: nothing reached the kernel,
reply another way. unsupported: same, and this descriptor rejects
splice (EINVAL), so do not offer it again. sent: done. failed: a
submission error was already reported to the kernel; never reply
again. */
enum class FuseReplyResult { fallback, unsupported, sent, failed };
#endif
#include <stdint.h>
#include <sys/stat.h>
#include <sys/statvfs.h>
@@ -288,6 +299,31 @@ int fuse_reply_data(fuse_req_t *req,
char *buf,
size_t bufsize);
#ifdef __cplusplus
/* READ replies at or above this size are offered to the splice path.
Measured on Linux 5.14 (tmpfs branch, release build) the splice
path uses less daemon CPU than read+writev from roughly 48KiB per
request; 64KiB keeps a margin. */
constexpr size_t FUSE_SPLICE_REPLY_MIN = 65536;
/* Whether a READ reply for size bytes at offset would currently be
attempted by fuse_reply_data_splice_fd: false below
FUSE_SPLICE_REPLY_MIN, once splice is unavailable, or when the reply
needs a pipe larger than pipe-max-size allows. The high-level READ
handler asks read_buf for a region only when this holds, so replies
that could only fall back cost no work in the filesystem. */
bool fuse_reply_splice_fits(off_t offset, size_t size);
/* Reply to a READ with a region of a file, spliced into /dev/fuse
without the data crossing userspace (Linux only; elsewhere, and
whenever the splice path is unavailable, returns fallback). Borrows
req: the caller frees it after a sent/failed outcome and must reply
by other means after fallback. Releases region.lock once the data
is staged, before the reply can reach the kernel. */
FuseReplyResult fuse_reply_data_splice_fd(fuse_req_t *req,
FileRegion &region);
#endif
/**
* Reply with data vector
*
@@ -536,7 +572,8 @@ int fuse_session_loop_mt(struct fuse_session *se,
const int read_thread_count,
const int process_thread_count,
const int process_thread_queue_depth,
const char *pin_threads_type);
const char *pin_threads_type,
const uint64_t read_spin_us);
/* Direct file descriptor and buffer size accessors
* (channel fields inlined into session for single-mount simplicity) */
+1
View File
@@ -10,6 +10,7 @@ struct fuse_req_t
{
fuse_req_ctx_t ctx;
struct fuse_session *se;
int fd;
fuse_conn_info_t conn;
unsigned int ioctl_64bit : 1;
};
+2 -2
View File
@@ -1191,7 +1191,7 @@ fuse_syslog_fuse_init_in(const struct fuse_init_in *arg_)
output.pop_back();
output += ");";
SysLog::info(output);
SysLog::info("{}",output);
}
void
@@ -1239,7 +1239,7 @@ fuse_syslog_fuse_init_out(const struct fuse_init_out *arg_)
output.pop_back();
output += ");";
SysLog::info(output);
SysLog::info("{}",output);
}
+287 -107
View File
@@ -1064,6 +1064,29 @@ free_path_wrlock(uint64_t nodeid,
free(path);
}
static
void
update_stat(node_t *node_,
const struct stat *stnew_)
{
uint32_t crc32b;
// This function can only invalidate an existing fingerprint; it
// never arms a new one. Arming (writing a non-zero CRC into
// state_crc32b) happens exclusively in open_auto_cache(), which is
// the single point that decides a file's page cache is worth
// retaining across opens. If stat_crc32b is 0 here the cache is
// already considered invalid and we have nothing to compare
// against.
if(node_->stat_crc32b == 0)
return;
crc32b = stat_crc32b(stnew_);
if(crc32b != node_->stat_crc32b)
node_->stat_crc32b = 0;
}
static
void
free_path(uint64_t nodeid,
@@ -1073,6 +1096,59 @@ free_path(uint64_t nodeid,
free_path_wrlock(nodeid,NULL,path);
}
static
node_t*
free_path_and_inc_open(u64 free_nodeid_,
char *path_,
u64 inc_open_ino_,
bool remember_too_)
{
node_t *node;
mutex_lock(f.lock);
unlock_path(free_nodeid_,NULL,NULL);
node = get_node(inc_open_ino_);
if(node)
{
if(remember_too_ && remember_nodes())
remember_node(node);
node->open_count++;
}
if(f.lockq)
wake_up_queued();
mutex_unlock(f.lock);
free(path_);
return node;
}
static
bool
free_path_and_update_stat(u64 nodeid_,
char *path_,
const struct stat *stnew_)
{
bool ok;
node_t *node;
mutex_lock(f.lock);
unlock_path(nodeid_,NULL,NULL);
node = get_node(nodeid_);
ok = (node != NULL);
if(ok)
update_stat(node,stnew_);
if(f.lockq)
wake_up_queued();
mutex_unlock(f.lock);
free(path_);
return ok;
}
static
void
free_path2(uint64_t nodeid1,
@@ -1208,52 +1284,65 @@ node_open(const node_t *node_)
return ((node_ != NULL) && (node_->open_count > 0));
}
static
void
update_stat(node_t *node_,
const struct stat *stnew_)
{
uint32_t crc32b;
// This function can only invalidate an existing fingerprint; it
// never arms a new one. Arming (writing a non-zero CRC into
// state_crc32b) happens exclusively in open_auto_cache(), which is
// the single point that decides a file's page cache is worth
// retaining across opens. If stat_crc32b is 0 here the cache is
// already considered invalid and we have nothing to compare
// against.
if(node_->stat_crc32b == 0)
return;
crc32b = stat_crc32b(stnew_);
if(crc32b != node_->stat_crc32b)
node_->stat_crc32b = 0;
}
static
int
set_path_info(uint64_t nodeid,
const char *name,
struct fuse_entry_param *e,
bool remember = true)
set_path_info(uint64_t nodeid_,
const char *name_,
struct fuse_entry_param *e_,
bool remember_ = true)
{
node_t *node;
node = find_node(nodeid,name,remember);
if(node == NULL)
return -ENOMEM;
e->ino = node->nodeid;
e->generation = ((e->ino == FUSE_ROOT_ID) ? 0 : f.nodeid_gen.generation);
mutex_lock(f.lock);
update_stat(node,&e->attr);
if(!name_)
{
node = get_node(nodeid_);
if(node == NULL)
goto out_unlock;
}
else
{
node = lookup_node(nodeid_,name_);
}
if(node == NULL)
{
node = node_alloc();
if(node == NULL)
goto out_unlock;
node->nodeid = generate_nodeid(&f.nodeid_gen);
if(remember_ && remember_nodes())
remember_node(node);
if(hash_name(node,nodeid_,name_) == -1)
{
free_node(node);
node = NULL;
goto out_unlock;
}
hash_id(node);
}
inc_nlookup(node);
e_->ino = node->nodeid;
e_->generation = ((e_->ino == FUSE_ROOT_ID) ? 0 : f.nodeid_gen.generation);
update_stat(node,&e_->attr);
mutex_unlock(f.lock);
set_stat(e->ino,&e->attr);
set_stat(e_->ino,&e_->attr);
return 0;
out_unlock:
mutex_unlock(f.lock);
return -ENOMEM;
}
/*
@@ -1474,26 +1563,37 @@ fuse_lib_getattr(fuse_req_t *req_,
if(!err)
{
bool node_ok = true;
if(fusepath != NULL)
err = f.ops.getattr(&req_->ctx,&fusepath[1],&buf,&timeouts);
{
err = f.ops.getattr(&req_->ctx,&fusepath[1],&buf,&timeouts);
if(!err)
node_ok = free_path_and_update_stat(hdr_->nodeid,fusepath,&buf);
else
free_path(hdr_->nodeid,fusepath);
}
else
err = f.ops.fgetattr(&req_->ctx,fh,&buf,&timeouts);
{
err = f.ops.fgetattr(&req_->ctx,fh,&buf,&timeouts);
if(!err)
{
mutex_lockguard(f.lock);
node = get_node(hdr_->nodeid);
node_ok = (node != NULL);
if(node_ok)
update_stat(node,&buf);
}
}
free_path(hdr_->nodeid,fusepath);
if(!err && !node_ok)
{
fuse_reply_err(req_,ESTALE);
return;
}
}
if(!err)
{
{
mutex_lockguard(f.lock);
node = get_node(hdr_->nodeid);
if(node == NULL)
{
fuse_reply_err(req_,ESTALE);
return;
}
update_stat(node,&buf);
}
set_stat(hdr_->nodeid,&buf);
fuse_reply_attr(req_,&buf,timeouts.attr);
}
@@ -1663,29 +1763,51 @@ fuse_lib_setattr(fuse_req_t *req_,
f.ops.getattr(&req_->ctx,&fusepath[1],&stbuf,&timeouts) :
f.ops.fgetattr(&req_->ctx,fh,&stbuf,&timeouts));
free_path(hdr_->nodeid,fusepath);
bool stat_updated = false;
if(!err && (fusepath != NULL))
{
node_t *node;
mutex_lock(f.lock);
unlock_path(hdr_->nodeid,NULL,NULL);
node = get_node(hdr_->nodeid);
if(node)
update_stat(node,&stbuf);
if(f.lockq)
wake_up_queued();
mutex_unlock(f.lock);
free(fusepath);
stat_updated = true;
}
else
{
free_path(hdr_->nodeid,fusepath);
}
if(!err)
{
if(!stat_updated)
{
node_t *node;
mutex_lockguard(f.lock);
node = get_node(hdr_->nodeid);
if(node)
update_stat(node,&stbuf);
}
set_stat(hdr_->nodeid,&stbuf);
fuse_reply_attr(req_,&stbuf,timeouts.attr);
}
else
{
fuse_reply_err(req_,err);
}
return;
}
if(!err)
{
{
node_t *node;
mutex_lockguard(f.lock);
node = get_node(hdr_->nodeid);
if(node)
update_stat(node,&stbuf);
}
set_stat(hdr_->nodeid,&stbuf);
fuse_reply_attr(req_,&stbuf,timeouts.attr);
}
else
{
fuse_reply_err(req_,err);
}
fuse_reply_err(req_,err);
}
static
void
fuse_lib_access(fuse_req_t *req_,
@@ -2089,15 +2211,8 @@ fuse_lib_create(fuse_req_t *req_,
if(!err)
{
mutex_lock(f.lock);
node_t *node = get_node(e.ino);
if(node)
{
if(remember_nodes())
remember_node(node);
node->open_count++;
}
mutex_unlock(f.lock);
free_path_and_inc_open(hdr_->nodeid,fusepath,e.ino,true);
fusepath = NULL;
if(fuse_reply_create(req_,&e,&ffi) == -ENOENT)
{
@@ -2196,11 +2311,9 @@ fuse_lib_open(fuse_req_t *req_,
if(!err)
{
mutex_lock(f.lock);
node_t *node = get_node(hdr_->nodeid);
if(node)
node->open_count++;
mutex_unlock(f.lock);
free_path_and_inc_open(hdr_->nodeid,fusepath,hdr_->nodeid,false);
fusepath = NULL;
/* The open syscall was interrupted,so it must be cancelled */
if(fuse_reply_open(req_,&ffi) == -ENOENT)
fuse_do_release(&req_->ctx,hdr_->nodeid,&ffi);
@@ -2215,22 +2328,14 @@ fuse_lib_open(fuse_req_t *req_,
static
void
fuse_lib_read(fuse_req_t *req_,
struct fuse_in_header *hdr_)
fuse_lib_read_mem(fuse_req_t *req_,
const fuse_file_info_t *ffi_,
const size_t size_,
const off_t offset_)
{
int res;
fuse_file_info_t ffi = {};
struct fuse_read_in *arg;
fuse_msgbuf_t *msgbuf;
arg = (fuse_read_in*)fuse_hdr_arg(hdr_);
ffi.fh = arg->fh;
if(req_->conn.proto_minor >= 9)
{
ffi.flags = arg->flags;
ffi.lock_owner = arg->lock_owner;
}
msgbuf = msgbuf_alloc_page_aligned();
if(msgbuf == NULL)
{
@@ -2239,10 +2344,10 @@ fuse_lib_read(fuse_req_t *req_,
}
res = f.ops.read(&req_->ctx,
&ffi,
ffi_,
msgbuf->mem,
arg->size,
arg->offset);
size_,
offset_);
if(res >= 0)
fuse_reply_data(req_,msgbuf->mem,res);
@@ -2252,6 +2357,84 @@ fuse_lib_read(fuse_req_t *req_,
msgbuf_free(msgbuf);
}
/* Large reads: ask the filesystem for a branch-file region and splice
it into /dev/fuse. Every staging failure falls back to the memory
path above, which is always correct. A failure of the final
submission splice is answered with an error-only reply and never
replayed: the kernel may already have consumed the payload. The
region's lock is released by the reply path once the data is staged;
after sent/failed only libfuse's own req is touched here, since the
handle may already be released. */
static
bool
fuse_lib_read_splice(fuse_req_t *req_,
const fuse_file_info_t *ffi_,
const size_t size_,
const off_t offset_)
{
int res;
FileRegion region;
res = f.ops.read_buf(&req_->ctx,ffi_,&region,size_,offset_);
if(res < 0)
{
fuse_reply_err(req_,res);
return true;
}
if(res > 0)
return false;
/* The reply length comes from region.size alone (see read_buf in
fuse.h): a region smaller than the request would be sent as a short
reply, which the kernel treats as end-of-file. Decline such a region
instead; its destructor releases the lock before the copy path. */
if(region.size != size_)
return false;
switch(fuse_reply_data_splice_fd(req_,region))
{
case FuseReplyResult::sent:
case FuseReplyResult::failed:
fuse_req_free(req_);
return true;
case FuseReplyResult::unsupported:
if(f.ops.read_buf_rejected)
f.ops.read_buf_rejected(&req_->ctx,ffi_);
return false;
case FuseReplyResult::fallback:
break;
}
return false;
}
static
void
fuse_lib_read(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
fuse_file_info_t ffi = {};
struct fuse_read_in *arg;
arg = (fuse_read_in*)fuse_hdr_arg(hdr_);
ffi.fh = arg->fh;
if(req_->conn.proto_minor >= 9)
{
ffi.flags = arg->flags;
ffi.lock_owner = arg->lock_owner;
}
/* A region is always the whole request, so whether its reply can be
spliced at all (the threshold, the pipe limit) is known here,
before the filesystem does any work for it. */
if((f.ops.read_buf != nullptr) &&
fuse_reply_splice_fits(arg->offset,arg->size) &&
fuse_lib_read_splice(req_,&ffi,arg->size,arg->offset))
return;
fuse_lib_read_mem(req_,&ffi,arg->size,arg->offset);
}
static
void
fuse_lib_write(fuse_req_t *req_,
@@ -2543,7 +2726,11 @@ fuse_lib_releasedir(fuse_req_t *req_,
f.ops.releasedir(&req_->ctx,
&ffi);
/* Done to keep race condition between last readdir reply and the unlock */
// Acquire/release barrier: not a real critical section. A
// concurrent readdir may still hold dh->lock at this point. This
// lock+unlock pair blocks until the last readdir drops the lock
// guaranteeing that the subsequent destroy/free sees no in-flight
// readdir accessing dh->d. Do NOT remove.
mutex_lock(dh->lock);
mutex_unlock(dh->lock);
mutex_destroy(dh->lock);
@@ -2924,15 +3111,8 @@ fuse_lib_tmpfile(fuse_req_t *req_,
if(!err)
{
mutex_lock(f.lock);
node_t *node = get_node(e.ino);
if(node)
{
if(remember_nodes())
remember_node(node);
node->open_count++;
}
mutex_unlock(f.lock);
free_path_and_inc_open(hdr_->nodeid,fusepath,e.ino,true);
fusepath = NULL;
if(fuse_reply_create(req_,&e,&ffi) == -ENOENT)
{
+15 -5
View File
@@ -1,5 +1,7 @@
#include "fuse_cfg.hpp"
#include <mutex>
fuse_cfg_t fuse_cfg;
bool
@@ -23,24 +25,32 @@ fuse_cfg_t::valid_umask() const
std::shared_ptr<FILE>
fuse_cfg_t::log_file() const
{
return std::atomic_load(&_log_file);
std::shared_lock<std::shared_mutex> lk(_log_mutex);
return _log_file;
}
void
fuse_cfg_t::log_file(std::shared_ptr<FILE> f_)
{
std::atomic_store(&_log_file, f_);
std::unique_lock<std::shared_mutex> lk(_log_mutex);
_log_file = std::move(f_);
}
std::shared_ptr<std::string>
fuse_cfg_t::log_filepath() const
{
return std::atomic_load(&_log_filepath);
std::shared_lock<std::shared_mutex> lk(_log_mutex);
return _log_filepath;
}
void
fuse_cfg_t::log_filepath(const std::string &s_)
{
std::atomic_store(&_log_filepath,
std::make_shared<std::string>(s_));
auto filepath = std::make_shared<std::string>(s_);
std::unique_lock<std::shared_mutex> lk(_log_mutex);
_log_filepath = std::move(filepath);
}
+13 -2
View File
@@ -15,6 +15,7 @@
#include "extern_c.h"
#include <atomic>
#include <vector>
/* Simplified fuse_session - fields inlined from former fuse_chan
* Since mergerfs only supports one mount, we collapse the hierarchy:
@@ -24,9 +25,11 @@
struct fuse_session
{
int (*receive_buf)(struct fuse_session *se,
int fd,
fuse_msgbuf_t *msgbuf);
void (*process_buf)(struct fuse_session *se,
int fd,
const fuse_msgbuf_t *msgbuf);
void (*destroy)(void *data);
@@ -35,8 +38,10 @@ struct fuse_session
std::atomic<int> exited;
/* Formerly in fuse_chan - inlined for single-mount simplicity */
int fd; /* /dev/fuse file descriptor */
size_t bufsize; /* Buffer size for I/O operations */
int fd; /* /dev/fuse file descriptor */
size_t bufsize; /* Buffer size for I/O operations */
std::vector<int> clone_fds; /* cloned /dev/fuse fds for read workers */
};
struct fuse_notify_req
@@ -50,6 +55,12 @@ struct fuse_notify_req
struct fuse_notify_req *prev;
};
bool fuse_session_setup_read_fds(struct fuse_session *se,
int read_thread_count);
int fuse_session_read_fd(struct fuse_session *se,
int index);
EXTERN_C_BEGIN
+79 -21
View File
@@ -11,7 +11,6 @@
#include "maintenance_thread.hpp"
#include "fuse_i.hpp"
#include "fuse_kernel.h"
#include "fuse_lowlevel.h"
#include "fuse_cfg.hpp"
@@ -22,6 +21,8 @@
#include <vector>
#include <errno.h>
#include <poll.h>
#include <time.h>
#include <pthread.h>
#include <semaphore.h>
#include <signal.h>
@@ -67,13 +68,16 @@ _nproc()
struct AsyncWorker
{
fuse_session *_se;
int _fd;
sem_t *_finished;
std::shared_ptr<ThreadPool> _process_tp;
AsyncWorker(fuse_session *se_,
int fd_,
sem_t *finished_,
std::shared_ptr<ThreadPool> process_tp_)
: _se(se_),
_fd(fd_),
_finished(finished_),
_process_tp(process_tp_)
{
@@ -102,7 +106,7 @@ struct AsyncWorker
while(true)
{
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);
rv = _se->receive_buf(_se,msgbuf);
rv = _se->receive_buf(_se,_fd,msgbuf);
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL);
if(rv > 0)
@@ -118,9 +122,9 @@ struct AsyncWorker
}
_process_tp->enqueue_work(ptok,
[=]()
[se = _se,fd = _fd,msgbuf]()
{
_se->process_buf(_se,msgbuf);
se->process_buf(se,fd,msgbuf);
msgbuf_free(msgbuf);
});
}
@@ -130,13 +134,19 @@ struct AsyncWorker
struct SyncWorker
{
fuse_session *_se;
int _fd;
sem_t *_finished;
u64 _spin_us;
SyncWorker(fuse_session *se_,
sem_t *finished_)
int fd_,
sem_t *finished_,
u64 spin_us_)
: _se(se_),
_finished(finished_)
_fd(fd_),
_finished(finished_),
_spin_us(spin_us_)
{
}
@@ -147,6 +157,7 @@ struct SyncWorker
DEFER{ fuse_session_exit(_se); };
DEFER{ sem_post(_finished); };
bool processed_one = false;
while(!fuse_session_exited(_se))
{
int rv;
@@ -159,10 +170,39 @@ struct SyncWorker
return;
}
/* After processing a request, poll /dev/fuse for up to _spin_us
before blocking in receive. On synchronous request streams (one
dd, one metadata walk) this measurably lowers per-request latency
versus blocking immediately; the exact scheduler mechanism is not
established. Cost: up to _spin_us of CPU per worker after each
request, so a busy-but-quiet mount pays roughly a core; nothing on
an idle mount. 0 disables. Cancellation stays disabled during the
spin (it is only enabled around the blocking receive) and the spin
is bounded, so shutdown is unaffected. */
if(processed_one && (_spin_us > 0))
{
struct pollfd pfd = {_fd,POLLIN,0};
struct timespec ts;
uint64_t end;
::clock_gettime(CLOCK_MONOTONIC,&ts);
end = ((uint64_t)ts.tv_sec * 1000000000ULL +
(uint64_t)ts.tv_nsec +
(_spin_us * 1000ULL));
while(::poll(&pfd,1,0) == 0)
{
::clock_gettime(CLOCK_MONOTONIC,&ts);
if(((uint64_t)ts.tv_sec * 1000000000ULL +
(uint64_t)ts.tv_nsec) >= end)
break;
}
}
processed_one = true;
while(true)
{
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);
rv = _se->receive_buf(_se,msgbuf);
rv = _se->receive_buf(_se,_fd,msgbuf);
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL);
if(rv > 0)
break;
@@ -176,7 +216,7 @@ struct SyncWorker
return ::_print_error(rv);
}
_se->process_buf(_se,msgbuf);
_se->process_buf(_se,_fd,msgbuf);
msgbuf_free(msgbuf);
}
@@ -252,18 +292,22 @@ fuse_session_loop_mt(struct fuse_session *se_,
const int raw_read_thread_count_,
const int raw_process_thread_count_,
const int raw_process_thread_queue_depth_,
const std::string pin_threads_type_)
const std::string pin_threads_type_,
const u64 read_spin_us_)
{
sem_t finished;
bool cloned_read_fds;
int read_thread_count;
int process_thread_count;
int process_thread_queue_depth;
std::vector<pthread_t> read_threads;
std::vector<pthread_t> process_threads;
std::unique_ptr<ThreadPool> read_tp;
std::shared_ptr<ThreadPool> process_tp;
sem_init(&finished,0,0);
DEFER{ sem_destroy(&finished); };
std::unique_ptr<ThreadPool> read_tp;
std::shared_ptr<ThreadPool> process_tp;
read_thread_count = raw_read_thread_count_;
process_thread_count = raw_process_thread_count_;
@@ -274,21 +318,29 @@ fuse_session_loop_mt(struct fuse_session *se_,
if(process_thread_count > 0)
process_tp = std::make_shared<ThreadPool>(process_thread_count,
process_thread_queue_depth,
"fuse.process");
process_thread_queue_depth,
"fuse.process");
cloned_read_fds = fuse_session_setup_read_fds(se_,read_thread_count);
read_tp = std::make_unique<ThreadPool>(read_thread_count,
read_thread_count,
"fuse.read");
read_thread_count,
"fuse.read");
if(process_tp)
{
for(auto i = 0; i < read_thread_count; i++)
read_tp->enqueue_work(AsyncWorker(se_,&finished,process_tp));
read_tp->enqueue_work(AsyncWorker(se_,
fuse_session_read_fd(se_,i),
&finished,
process_tp));
}
else
{
for(auto i = 0; i < read_thread_count; i++)
read_tp->enqueue_work(SyncWorker(se_,&finished));
read_tp->enqueue_work(SyncWorker(se_,
fuse_session_read_fd(se_,i),
&finished,
read_spin_us_));
}
if(read_tp)
@@ -301,11 +353,15 @@ fuse_session_loop_mt(struct fuse_session *se_,
SysLog::info("read-thread-count={}; "
"process-thread-count={}; "
"process-thread-queue-depth={}; "
"pin-threads={};",
"fuse-dev-ioc-clone={}; "
"pin-threads={}; "
"read-spin-us={};",
read_thread_count,
process_thread_count,
process_thread_queue_depth,
pin_threads_type_);
cloned_read_fds,
pin_threads_type_,
read_spin_us_);
while(!fuse_session_exited(se_))
sem_wait(&finished);
@@ -313,7 +369,8 @@ fuse_session_loop_mt(struct fuse_session *se_,
for(auto t : read_threads)
pthread_cancel(t);
sem_destroy(&finished);
read_tp.reset();
process_tp.reset();
return 0;
}
@@ -333,7 +390,8 @@ fuse_loop_mt(struct fuse *f_)
fuse_cfg.read_thread_count,
fuse_cfg.process_thread_count,
fuse_cfg.process_thread_queue_depth,
fuse_cfg.pin_threads);
fuse_cfg.pin_threads,
fuse_cfg.read_spin_us);
MaintenanceThread::stop();
+518 -14
View File
@@ -22,6 +22,14 @@
#include "fuse_opt.h"
#include "fuse_pollhandle.h"
#include "stat_utils.h"
#include "syslog.hpp"
#include <atomic>
#if defined(__linux__)
#include <fcntl.h>
#include <sys/ioctl.h>
#endif
#include <stdio.h>
#include <stdlib.h>
@@ -103,16 +111,16 @@ iov_length(const struct iovec *iov,
static
int
fuse_send_msg(struct fuse_session *se,
struct iovec *iov,
int count)
fuse_send_msg(const int fd_,
struct iovec *iov,
int count)
{
int rv;
struct fuse_out_header *out = (fuse_out_header*)iov[0].iov_base;
out->len = iov_length(iov, count);
rv = writev(se->fd,iov,count);
rv = writev(fd_,iov,count);
if(rv == -1)
return -errno;
@@ -144,7 +152,7 @@ fuse_send_reply_iov_nofree(fuse_req_t *req,
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(struct fuse_out_header);
return fuse_send_msg(req->se, iov, count);
return fuse_send_msg(req->fd, iov, count);
}
static
@@ -420,6 +428,493 @@ fuse_reply_buf(fuse_req_t *req,
return send_reply_ok(req, buf, size);
}
#if defined(__linux__)
/* Automatic splice of large READ replies (branch fd -> pipe -> /dev/fuse).
The copy path costs read + pread + writev and copies the payload
twice. The splice path costs read + write(header) + splice(fd ->
pipe) + splice(pipe -> /dev/fuse) and never copies the payload into
userspace; the kernel copies it once, into the request's pages. The
caller gates on fuse_reply_splice_fits (fuse_lowlevel.h).
Every staging failure is recoverable: nothing has reached the kernel,
so the pipe is drained and the caller replies through the copy path.
Only a kernel without splice (ENOSYS), or a seccomp policy refusing
it, disables the path for the whole process. A pipe-max-size too
small for a reply only caps the reply size that is spliced
(g_reply_pipe_limit), and a branch that rejects splice is reported
per open file (FuseReplyResult::unsupported). */
static std::atomic<bool> g_reply_splice_off{false};
/* Smallest reply pipe, in bytes, found to exceed the pipe-max-size
sysctl, which caps unprivileged F_SETPIPE_SZ; a 1MiB reply plus its
header page needs one page more than the default 1MiB. Replies that
need at least this much stay on the copy path unless this thread's
pipe already holds them; smaller replies keep splicing. A refusal
from the per-user pipe quota is not recorded: it lifts as pipes are
closed. 0 = no refusal seen. */
static std::atomic<size_t> g_reply_pipe_limit{0};
/* The pipe-max-size sysctl, read once at session setup. Read lazily, on
the first refused resize, it could fail for the same shortage of
descriptors that made that resize or pipe fail and then read as 0 for
good. 0 when it cannot be read, in which case every EPERM is treated
as that limit. */
static std::atomic<size_t> g_pipe_max_size{0};
static
void
_read_pipe_max_size(void)
{
unsigned long n = 0;
FILE *f = ::fopen("/proc/sys/fs/pipe-max-size","re");
if(f == NULL)
return;
if(::fscanf(f,"%lu",&n) == 1)
g_pipe_max_size.store((size_t)n,std::memory_order_relaxed);
::fclose(f);
}
static
void
_reply_splice_disable(const char *what_,
const int err_)
{
if(g_reply_splice_off.exchange(true))
return;
SysLog::notice("read reply splice disabled: {} failed: {};"
" using copy path",
what_,strerror(err_));
}
/* One latch per distinct failure reason: these are independent,
recurring conditions (a pipe quota hiccup, a header-staging error, a
branch-splice error, a submission error, ...), so silencing all of
them after whichever happens to fire first would hide later,
unrelated failures from the log. A submission the kernel refused
before it looked at the request (the copy path then answers it) is
a different incident from one that fails the request, so they do
not share a latch. */
enum class SpliceWarnReason
{
pipe_quota,
pipe_create,
header_stage,
branch_splice,
submit_fallback,
submit_failed,
_count
};
static std::atomic<bool> g_reply_splice_warned[(size_t)SpliceWarnReason::_count] = {};
static
void
_reply_splice_warn(const char *what_,
const int err_,
const SpliceWarnReason reason_,
const char *outcome_ =
"falling back to copy path for this reply")
{
if(g_reply_splice_warned[(size_t)reason_].exchange(true))
return;
SysLog::warning("read reply splice: {} failed: {}; {}",
what_,strerror(err_),outcome_);
}
/* pipe_quota (the per-user pipe count/memory limit) is the one reason
that is genuinely transient -- it lifts as pipes are closed, unlike
the others, which are properties of the kernel or a branch fd that
do not change at runtime. Clear its latch once a resize succeeds, so
a later recurrence (a distinct incident) is logged again instead of
being silenced forever by whichever incident fired first. */
static
void
_reply_splice_warn_clear(const SpliceWarnReason reason_)
{
g_reply_splice_warned[(size_t)reason_].store(false,std::memory_order_relaxed);
}
/* Whether splice itself is unusable in this process: missing from the
kernel (ENOSYS) or refused by a seccomp policy, which commonly
answers EPERM but may use any errno. A zero-length splice returns 0
before looking at its descriptors, so only then does it fail. An
error from splicing a branch file normally comes from that file's
own read path instead (ENOSYS from a FUSE branch whose server does
not implement READ, EIO, ...), which this tells apart. */
static
bool
_splice_syscall_unusable()
{
return (::splice(-1,nullptr,-1,nullptr,0,0) == -1);
}
static
void
fuse_send_errno(const int fd_,
const int errno_,
const uint64_t unique_id_);
namespace
{
/* One pipe per worker thread, created on first use and reused for
every large reply that thread sends. Empty between replies: every
path that leaves bytes behind drains or closes it. */
struct ReplyPipe
{
int fd[2] = {-1,-1};
size_t cap = 0;
~ReplyPipe() { reset(); }
void
reset()
{
if(fd[0] != -1) ::close(fd[0]);
if(fd[1] != -1) ::close(fd[1]);
fd[0] = fd[1] = -1;
cap = 0;
}
/* 0 when a pipe of at least bytes_ capacity is held, else -errno. */
int
ready(const size_t bytes_)
{
if(fd[0] == -1)
{
/* The initial capacity is not queried: cap 0 makes the
first use resize to exactly what it needs, which with 4KiB
pages it would have to do anyway (a reply of 64KiB plus
its header needs more than the default 16 slots). */
if(::pipe2(fd,O_CLOEXEC|O_NONBLOCK) == -1)
return -errno;
}
if(bytes_ <= cap)
return 0;
if(bytes_ > (size_t)INT_MAX)
return -EINVAL;
const int c = ::fcntl(fd[0],F_SETPIPE_SZ,(int)bytes_);
if(c == -1)
return -errno;
cap = (size_t)c;
return ((cap >= bytes_) ? 0 : -ENOSPC);
}
/* /dev/null, opened on first use and kept. A failed open (EMFILE at
that moment, say) is retried by the next drain rather than
cached, which would leave every later drain copying. */
static
int
devnull()
{
static std::atomic<int> g_fd{-1};
int fd = g_fd.load(std::memory_order_acquire);
if(fd != -1)
return fd;
fd = ::open("/dev/null",O_WRONLY|O_CLOEXEC);
if(fd == -1)
return -1;
int expected = -1;
if(g_fd.compare_exchange_strong(expected,fd,std::memory_order_acq_rel))
return fd;
::close(fd);
return expected;
}
/* Discard anything staged. Splicing it into /dev/null drops the
page references without copying the data into userspace; reading
it out is the fallback when that is not available (no /dev/null,
a buffer error), so the pipe/capacity survive to be reused. Both
ends are non-blocking, so each stops at empty (EAGAIN); a pipe
that cannot be emptied by either is closed instead, and ready()
makes a new one on next use. */
void
drain()
{
/* true once the pipe reads empty (EAGAIN), false on any other
outcome. */
const auto pump = [](auto op_)
{
for(;;)
{
const ssize_t rv = op_();
if((rv > 0) || ((rv == -1) && (errno == EINTR)))
continue;
return ((rv == -1) && (errno == EAGAIN));
}
};
const int devnull = ReplyPipe::devnull();
if((devnull != -1) &&
pump([&]{ return ::splice(fd[0],nullptr,devnull,nullptr,
(size_t)1 << 30,SPLICE_F_NONBLOCK); }))
return;
char buf[16384];
if(pump([&]{ return ::read(fd[0],buf,sizeof(buf)); }))
return;
reset();
}
};
}
static thread_local ReplyPipe t_reply_pipe;
/* Pipe capacity a reply needs: one slot for the header plus the file
range, which may start mid-page. The kernel fills pipes one page per
slot, so capacity is counted in pages, not bytes. */
static
size_t
_reply_pipe_need(const off_t offset_,
const size_t size_)
{
const size_t slots =
(((uintptr_t)offset_ % pagesize) + size_ + pagesize - 1) / pagesize;
return (1 + slots) * pagesize;
}
/* Whether a pipe of need_ bytes may be used: always below any recorded
pipe-max-size refusal, and above it only when this thread's pipe
already has that capacity (so no resize is attempted). */
static
bool
_reply_pipe_allowed(const size_t need_)
{
const size_t limit = g_reply_pipe_limit.load(std::memory_order_relaxed);
return ((limit == 0) || (need_ < limit) || (need_ <= t_reply_pipe.cap));
}
/* The pipe capacity a reply of size_ bytes at offset_ needs, or 0 when
it is not spliced: splice is off, the reply is below the threshold
(cheaper through read + writev), or it needs a pipe over a recorded
pipe-max-size refusal. The one predicate behind both
fuse_reply_splice_fits and fuse_reply_data_splice_fd. */
static
size_t
_reply_splice_need(const off_t offset_,
const size_t size_)
{
if(g_reply_splice_off.load(std::memory_order_relaxed) ||
(size_ < FUSE_SPLICE_REPLY_MIN))
return 0;
const size_t need = _reply_pipe_need(offset_,size_);
return (_reply_pipe_allowed(need) ? need : 0);
}
bool
fuse_reply_splice_fits(const off_t offset_,
const size_t size_)
{
return (_reply_splice_need(offset_,size_) != 0);
}
FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *req_,
FileRegion &region_)
{
const size_t need = _reply_splice_need(region_.offset,region_.size);
if(need == 0)
return FuseReplyResult::fallback;
ReplyPipe &p = t_reply_pipe;
struct fuse_out_header out = {};
out.unique = req_->ctx.unique;
out.len = sizeof(out) + region_.size;
const size_t prev_cap = p.cap;
int err = p.ready(need);
if(err != 0)
{
/* EPERM means either larger than pipe-max-size without
CAP_SYS_RESOURCE, a fixed bound worth remembering so replies
this large skip straight to the copy path, or over the
per-user pipe quota, which is transient: only this reply falls
back. An unreadable sysctl is treated as the fixed bound. */
const size_t max = g_pipe_max_size.load(std::memory_order_relaxed);
if((err == -EPERM) && ((max == 0) || (need > max)))
{
size_t cur = g_reply_pipe_limit.load(std::memory_order_relaxed);
while(((cur == 0) || (need < cur)) &&
!g_reply_pipe_limit.compare_exchange_weak(cur,need))
;
/* cur is still 0 only in the thread that set the first limit. */
if(cur == 0)
SysLog::notice("read reply splice: a {} byte pipe exceeds"
" pipe-max-size; reads this large use the"
" copy path, smaller reads are still spliced",
need);
}
else if(err == -EPERM)
_reply_splice_warn("growing the reply pipe (per-user pipe quota)",-err,
SpliceWarnReason::pipe_quota);
else if(err == -ENOSYS)
_reply_splice_disable("creating the reply pipe",-err);
else
_reply_splice_warn("creating the reply pipe",-err,
SpliceWarnReason::pipe_create);
return FuseReplyResult::fallback;
}
/* A resize just succeeded, so any earlier per-user pipe quota
shortage (the one transient SpliceWarnReason) has lifted; let a
later recurrence be logged again instead of staying silenced by an
earlier one. A pipe that was already big enough says nothing about
the quota: it must not re-arm the warning while other threads still
fail to grow theirs, nor write the shared latch on every reply. */
if(p.cap != prev_cap)
_reply_splice_warn_clear(SpliceWarnReason::pipe_quota);
/* write(), not vmsplice(): vmsplice without SPLICE_F_GIFT queues a
reference to this stack page rather than a copy, and the pipe
outlives the frame whenever it is consumed later (or not at all).
A 16-byte write into an empty pipe copies and is atomic. */
ssize_t n;
do
n = ::write(p.fd[1],&out,sizeof(out));
while((n == -1) && (errno == EINTR));
if(n != (ssize_t)sizeof(out))
{
err = ((n == -1) ? errno : EIO);
p.drain();
_reply_splice_warn("staging the reply header",err,
SpliceWarnReason::header_stage);
return FuseReplyResult::fallback;
}
size_t done = 0;
while(done < region_.size)
{
loff_t pos = region_.offset + (off_t)done;
n = ::splice(region_.fd,&pos,p.fd[1],nullptr,
region_.size - done,SPLICE_F_NONBLOCK);
if((n == -1) && (errno == EINTR))
continue;
if(n > 0)
{
done += (size_t)n;
continue;
}
/* n == 0: the file ended before the region did (it shrank after
the caller sized the region). n == -1: this descriptor rejects
splice (EINVAL: its filesystem has no splice_read) or failed.
EAGAIN is also end-of-file when the file ended inside the
region's last page of an exactly full pipe (no room is left
for the splice that would return 0). Nothing reached the
kernel. A rejecting descriptor is a property of that branch
file, not of the transport, so the path stays on and the
caller is told to stop offering it; only a first splice can
show that, since one that already moved data proves the
descriptor splices. An error disables the path only when
splice itself is unusable (missing, or refused by seccomp),
not when it is the branch file's read error; otherwise every
large read would pay for a staging attempt that cannot
succeed. */
err = ((n == -1) ? errno : 0);
p.drain();
if((err != 0) && (err != EAGAIN))
{
if(_splice_syscall_unusable())
_reply_splice_disable("splicing from the branch file",err);
else if(err != EINVAL)
_reply_splice_warn("splicing from the branch file",err,
SpliceWarnReason::branch_splice);
}
return (((err == EINVAL) && (done == 0)) ?
FuseReplyResult::unsupported :
FuseReplyResult::fallback);
}
/* fd is not read again on any outcome below (see FileRegion::lock). */
if(region_.lock.owns_lock())
region_.lock.unlock();
/* The whole message is staged now, so a log write that throws (fmt on
a full or broken log file) must not leave it in this thread's pipe:
the worker survives the exception and would splice it in front of
its next reply. */
if(fuse_cfg.debug)
{
try
{
fuse_debug_data_out(req_->ctx.unique,region_.size);
}
catch(...)
{
p.drain();
throw;
}
}
/* Pages spliced in from the branch page cache are never offered for
SPLICE_F_MOVE: stealing them would evict the branch's cache pages.
The kernel copies them into the request's pages instead. */
const size_t total = out.len;
do
n = ::splice(p.fd[0],nullptr,req_->fd,nullptr,total,0);
while((n == -1) && (errno == EINTR));
if(n == (ssize_t)total)
return FuseReplyResult::sent;
err = ((n == -1) ? errno : EIO);
/* The kernel takes the whole message or none of it. If the pipe still
holds all of it, the submission failed before the kernel looked at
the request (e.g. ENOMEM allocating its buffer list), which is
still pending, so the copy path can answer it. */
int queued = 0;
if((::ioctl(p.fd[0],FIONREAD,&queued) == 0) && ((size_t)queued == total))
{
p.drain();
_reply_splice_warn("submitting the reply",err,
SpliceWarnReason::submit_fallback);
return FuseReplyResult::fallback;
}
/* Otherwise the kernel may have completed the request before failing
(it validates after copying; a copy error fails it with EIO), so
the payload is never replayed. ENOENT means the request was already
retired or the connection aborted: nothing is left to answer, as
when the copy path's writev fails the same way, so it is neither
logged nor answered. Any other error-only reply completes a request
that is still pending. */
p.drain();
if(err != ENOENT)
{
_reply_splice_warn("submitting the reply",err,
SpliceWarnReason::submit_failed,
"failing the request");
fuse_send_errno(req_->fd,err,req_->ctx.unique);
}
return FuseReplyResult::failed;
}
#else
bool
fuse_reply_splice_fits(const off_t,
const size_t)
{
return false;
}
FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *,
FileRegion &)
{
return FuseReplyResult::fallback;
}
#endif
int
fuse_reply_data(fuse_req_t *req,
char *buf_,
@@ -440,7 +935,7 @@ fuse_reply_data(fuse_req_t *req,
if(fuse_cfg.debug)
fuse_debug_data_out(req->ctx.unique,bufsize_);
res = fuse_send_msg(req->se,iov,2);
res = fuse_send_msg(req->fd,iov,2);
fuse_req_free(req);
return res;
@@ -1310,7 +1805,7 @@ send_notify_iov(struct fuse_session *se,
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(struct fuse_out_header);
return fuse_send_msg(se, iov, count);
return fuse_send_msg(se->fd, iov, count);
}
int
@@ -1558,7 +2053,7 @@ fuse_ll_destroy(void *data)
static
void
fuse_send_errno(struct fuse_session *se_,
fuse_send_errno(const int fd_,
const int errno_,
const uint64_t unique_id_)
{
@@ -1574,25 +2069,26 @@ fuse_send_errno(struct fuse_session *se_,
if(fuse_cfg.debug)
fuse_debug_out_header(&out);
fuse_send_msg(se_,&iov,1);
fuse_send_msg(fd_,&iov,1);
}
static
void
fuse_send_enomem(struct fuse_session *se_,
fuse_send_enomem(const int fd_,
const uint64_t unique_id_)
{
fuse_send_errno(se_,ENOMEM,unique_id_);
fuse_send_errno(fd_,ENOMEM,unique_id_);
}
static
int
fuse_ll_buf_receive_read(struct fuse_session *se_,
int fd_,
fuse_msgbuf_t *msgbuf_)
{
int rv;
rv = read(se_->fd,msgbuf_->mem,msgbuf_->size);
rv = read(fd_,msgbuf_->mem,msgbuf_->size);
if(rv == -1)
return -errno;
@@ -1608,6 +2104,7 @@ fuse_ll_buf_receive_read(struct fuse_session *se_,
static
void
fuse_ll_buf_process_read(struct fuse_session *se_,
int fd_,
const fuse_msgbuf_t *msgbuf_)
{
int err;
@@ -1621,7 +2118,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_,
req = fuse_req_alloc();
if(req == NULL)
return fuse_send_enomem(se_,in->unique);
return fuse_send_enomem(fd_,in->unique);
req->ctx.len = in->len;
req->ctx.opcode = in->opcode;
@@ -1633,6 +2130,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_,
req->ctx.umask = 0;
req->conn = f.conn;
req->se = se_;
req->fd = fd_;
req->ioctl_64bit = 0;
err = ENOSYS;
@@ -1653,6 +2151,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_,
static
void
fuse_ll_buf_process_read_init(struct fuse_session *se_,
int fd_,
const fuse_msgbuf_t *msgbuf_)
{
int err;
@@ -1666,7 +2165,7 @@ fuse_ll_buf_process_read_init(struct fuse_session *se_,
req = fuse_req_alloc();
if(req == NULL)
return fuse_send_enomem(se_,in->unique);
return fuse_send_enomem(fd_,in->unique);
req->ctx.len = in->len;
req->ctx.opcode = in->opcode;
@@ -1678,6 +2177,7 @@ fuse_ll_buf_process_read_init(struct fuse_session *se_,
req->ctx.pid = in->pid;
req->ctx.umask = 0;
req->se = se_;
req->fd = fd_;
err = EIO;
if(in->opcode != FUSE_INIT)
@@ -1721,6 +2221,10 @@ fuse_lowlevel_new_common(struct fuse_args *args,
memcpy(&f.op,op,op_size);
f.owner = getuid();
#if defined(__linux__)
_read_pipe_max_size();
#endif
se = fuse_session_new(&f,
(void*)fuse_ll_buf_receive_read,
(void*)fuse_ll_buf_process_read_init,
+96 -2
View File
@@ -8,16 +8,24 @@
#include "fuse_i.hpp"
#include "fuse_kernel.h"
#include "syslog.hpp"
#include <cassert>
#include <cerrno>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <new>
#include <sys/ioctl.h>
#include <unistd.h>
#include <sys/uio.h>
#ifndef O_CLOEXEC
#define O_CLOEXEC 0
#endif
/*
* Simplified session management for single-mount filesystems.
* The fuse_chan structure has been removed and its fields inlined
@@ -25,6 +33,42 @@
* one mount point per process.
*/
static
int
_clone_fuse_fd(const int fd_)
{
int rv;
int clone_fd;
uint32_t source_fd;
clone_fd = ::open("/dev/fuse",O_RDWR|O_CLOEXEC);
if(clone_fd == -1)
return -errno;
source_fd = fd_;
rv = ::ioctl(clone_fd,FUSE_DEV_IOC_CLONE,&source_fd);
if(rv == -1)
{
int err = errno;
::close(clone_fd);
return -err;
}
return clone_fd;
}
static
void
_close_clone_fds(struct fuse_session *se_)
{
for(auto fd : se_->clone_fds)
{
if(fd != -1)
::close(fd);
}
se_->clone_fds.clear();
}
struct fuse_session*
fuse_session_new(void *data,
void *receive_buf,
@@ -40,8 +84,8 @@ fuse_session_new(void *data,
}
se->f = static_cast<struct fuse_ll*>(data);
se->receive_buf = reinterpret_cast<int(*)(struct fuse_session*, fuse_msgbuf_t*)>(receive_buf);
se->process_buf = reinterpret_cast<void(*)(struct fuse_session*, const fuse_msgbuf_t*)>(process_buf);
se->receive_buf = reinterpret_cast<int(*)(struct fuse_session*, int, fuse_msgbuf_t*)>(receive_buf);
se->process_buf = reinterpret_cast<void(*)(struct fuse_session*, int, const fuse_msgbuf_t*)>(process_buf);
se->destroy = reinterpret_cast<void(*)(void*)>(destroy);
se->fd = -1;
@@ -57,6 +101,8 @@ fuse_session_destroy(struct fuse_session *se)
if(se->destroy)
se->destroy(se->f);
::_close_clone_fds(se);
if(se->fd != -1)
{
close(se->fd);
@@ -93,6 +139,8 @@ fuse_session_data(struct fuse_session *se)
int
fuse_session_clearfd(struct fuse_session *se)
{
::_close_clone_fds(se);
int fd = se->fd;
se->fd = -1;
return fd;
@@ -102,9 +150,55 @@ void
fuse_session_setfd(struct fuse_session *se,
int fd)
{
::_close_clone_fds(se);
se->fd = fd;
}
bool
fuse_session_setup_read_fds(struct fuse_session *se_,
const int read_thread_count_)
{
::_close_clone_fds(se_);
if(read_thread_count_ <= 1)
return false;
se_->clone_fds.reserve(read_thread_count_ - 1);
for(auto i = 1; i < read_thread_count_; i++)
{
int fd;
fd = ::_clone_fuse_fd(se_->fd);
if(fd < 0)
{
int err = -fd;
::_close_clone_fds(se_);
SysLog::warning("failed to clone /dev/fuse fd for read threads; "
"using shared fd: {} ({})",
strerror(err),
err);
return false;
}
se_->clone_fds.push_back(fd);
}
return true;
}
int
fuse_session_read_fd(struct fuse_session *se_,
const int index_)
{
if((index_ > 0) && (index_ <= (int)se_->clone_fds.size()))
return se_->clone_fds[index_ - 1];
return se_->fd;
}
void
fuse_session_setbufsize(struct fuse_session *se,
size_t bufsize)