Restore splice zero-copy I/O as runtime options

Reinstate the splice read/write/move paths removed in 84592a9f, now
gated behind runtime options with negotiated pipe sizing and fallback
logging.

- Transport: msgbuf-piped splice receive, fd-referenced splice reply
  (vmsplice + splice), splice-move pipe-bounce, and restored
  fuse_buf_copy/fuse_buf_flags plumbing for copy fallbacks
- Negotiation: new pipe-max-size option (auto-raised at INIT when
  possible; the sysctl read/write is factored into src/pipe_max_size.cpp
  so fuse_init.cpp and option_parser.cpp share one implementation);
  fuse-msg-size clamped to what the pipe can hold; splice is disabled
  at INIT with a warning if the sysctl can't fit even the minimum,
  retracting the FUSE_CAP_SPLICE_* bits already advertised to the
  kernel in that case
- Read replies and splice-move are gated at 128 KiB (below that the
  syscalls cost more than the copy they avoid); every fallback path
  degrades to the copy path and logs once per process via SysLog
- msgbuf pipe cache is process-global and mutex-guarded (mutex_lockguard)
  so pipes can move between read and worker threads under
  process-thread mode
- API simplified after review: collapsed the three per-axis options
  (splice-read/write/move) into a single splice boolean (default on);
  no-splice-read/write/move and the bare splice-read/write/move
  spellings stay registered as accepted no-ops so existing mount
  options don't fail the mount, but they do NOT alter the splice
  knob - upgraders with no-splice-* in their config get splice
  enabled and must set splice=false explicitly (see
  deprecated_options.md)
- Fixed several correctness issues found in parallel review: a
  dropped write_buf registration that silently corrupted writes, a
  dirty-pipe reuse bug in msgbuf_free, an iov double-send in
  fuse_reply_data's splice-move fallback, and pipe-capacity math that
  assumed a 4096-byte page size
- Modernized the transport/negotiation code to this codebase's C++
  style: RAII (unique_ptr/DEFER) instead of malloc/free and manual fd
  close(), std::from_chars/to_chars instead of stoul/snprintf,
  _ReplyPipe's pipe ops as methods instead of free functions closing
  over a thread_local global, and vendored/libfuse/lib/buffer.cpp (a
  near-verbatim port of upstream's buffer.c) brought in line with the
  rest of the fork
- Documented splice and pipe-max-size options (mkdocs, FAQ,
  deprecated-options list); added tools/bench-splice.sh
- Benchmarked on tmpfs: seq_write +21%, seq_read +52%, small_write
  +5%, small_read +12%; validated with a 1h soak (~271 GiB
  hash-verified I/O, zero integrity failures), a live-mount round-trip
  of I/O above the splice threshold, and the existing unit/smoke
  suites at their pre-existing baseline

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Antonio S. Musumeci
2026-09-07 11:24:16 -04:00
co-authored by Claude Sonnet 5
parent 8ea47df13e
commit 8eed256747
30 changed files with 3098 additions and 42 deletions
+6
View File
@@ -19,6 +19,12 @@ function or have been replaced. **They should not be used.**
* **remember-nodes**: Use `never-forget-nodes`.
* **noforget**: Use `never-forget-nodes`.
* **nonempty**: Does nothing.
* **no-splice-read**: Does nothing. Use [splice=false](splice.md) to
disable splice.
* **no-splice-write**: Does nothing. Use [splice=false](splice.md) to
disable splice.
* **no-splice-move**: Does nothing. Use [splice=false](splice.md) to
disable splice.
* **splice_move**: Does nothing.
* **splice_read**: Does nothing.
* **splice_write**: Does nothing.
+10
View File
@@ -123,6 +123,16 @@ config file.
directory and symlink to it.
* **[readahead](readahead.md)=UINT**: Set readahead (in kilobytes) for
mergerfs and branches if greater than 0. (default: 0)
* **[splice](splice.md)=BOOL**: Enable the kernel's splice syscall to
move data between the FUSE device and the underlying filesystems
without copying through user space. Read replies of 128KiB or larger
are spliced; smaller read replies use the copy path. Writes stage
their payload through a pipe at any size. Falls back automatically
when unsupported. (default: true)
* **[pipe-max-size](pipe-max-size.md)=SIZE**: The value to write to
/proc/sys/fs/pipe-max-size at mount time, needed only when
fuse-msg-size exceeds the current limit while splice is enabled.
Root-only; unprivileged mounts log and clamp instead. (default: 0)
* **posix-acl=BOOL**: Enable POSIX ACL support (if supported by kernel
and underlying filesystem). (default: false)
* **async-read=BOOL**: Perform reads asynchronously. If disabled or
+34
View File
@@ -0,0 +1,34 @@
# pipe-max-size
* type: `SIZE`
* default: `0`
* example: `pipe-max-size=8M`
Sets the value written to `/proc/sys/fs/pipe-max-size` at mount time,
before the FUSE connection is negotiated. This is the kernel's limit
on the capacity of a single pipe, which bounds how much data can be
moved in a single `splice` call.
mergerfs already raises this sysctl on its own, automatically,
whenever splice is enabled (the default) and `fuse-msg-size` doesn't
fit the current `pipe-max-size` (default 1MiB on most kernels) - that
automatic behavior does not depend on this option being set. Setting
`pipe-max-size` explicitly is for cases where you want the sysctl
raised to a specific value up front (e.g. more headroom than the
minimum mergerfs would otherwise request), or want it raised before
mergerfs starts negotiating rather than during. If you use the
default `fuse-msg-size` you do not need this option.
Writing the sysctl requires running as root. When not root mergerfs
will log (at info level) that it could not raise the limit and will
clamp `fuse-msg-size` to what the current pipe limit allows instead.
Note that this is a system wide setting that persists after mergerfs
exits; mergerfs does not restore the previous value on unmount.
For unprivileged usage the kernel also caps the total number of pipe
pages a non-root user may have allocated, via
`/proc/sys/fs/pipe-user-pages-soft` and
`/proc/sys/fs/pipe-user-pages-hard` (both present since Linux
v4.5). This is a separate limit from `pipe-max-size`, which caps a
single pipe's capacity.
+33
View File
@@ -0,0 +1,33 @@
# splice
* type: `BOOL`
* default: `true`
* example: `splice=false`
Controls whether mergerfs uses the kernel's
[splice](https://man7.org/linux/man-pages/man2/splice.2.html) syscall
to move data between `/dev/fuse` and the underlying filesystems
without copying it through user space memory.
When enabled, `read` and `write` requests can be serviced with zero
copies: the kernel moves pages directly between the branch
filesystem's page cache and the FUSE device through pipes. This can
significantly improve sequential throughput on modern hardware.
Due to the extra syscall overhead for small requests, splice replies
to `read` requests are only used for requests of 128KiB or larger;
smaller read replies use the traditional write path. `write` requests
of any size stage their payload into the receive pipe and splice it
to the branch when possible.
If the filesystem or kernel does not support the needed
functionality mergerfs will automatically fall back to the
traditional read/write behavior.
Historical note: splice support was previously removed from
mergerfs's own vendored libfuse (see the
[FAQ](../faq/technical_behavior_and_limitations.md) for why), and
the `no-splice-read`, `no-splice-write`, and `no-splice-move`
spellings were accepted as no-ops during that period. They remain
accepted as no-ops for backward compatibility; use `splice=false`
instead.
@@ -215,13 +215,20 @@ If `mergerfs` doesn't work as a type it could be due to how the
permissions.
## Why was splice support removed?
## Why was splice support removed and then restored?
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.
Splice support was originally removed after testing showed it, at
best, provided equivalent performance and in some cases worse.
It was restored in 2026 after a rewrite for the current transport
layer: request receive staging, fd-referenced read replies, and a
128 KiB threshold below which read replies stay on the copy path
(write requests stage their payload through a pipe at any size). On
tmpfs branches the restored paths benchmark at seq_read +52%,
seq_write +21%. See [splice](../config/splice.md) and
[pipe-max-size](../config/pipe-max-size.md) in the options
documentation. Splice remains
unavailable on platforms that lack it; the copy path is the
fallback everywhere.
## How does mergerfs handle credentials?
+2
View File
@@ -84,6 +84,8 @@ nav:
- config/proxy-ioprio.md
- config/link-cow.md
- config/fuse-msg-size.md
- config/splice.md
- config/pipe-max-size.md
- config/follow-symlinks.md
- config/link-exdev.md
- config/rename-exdev.md
+36
View File
@@ -119,11 +119,32 @@ Config::Config()
proxy_ioprio(false),
read_thread_count(fuse_cfg.read_thread_count),
readahead(0),
/* pipe_max_size=0: no sysctl write (default). Non-zero: write value
to /proc/sys/fs/pipe-max-size BEFORE FUSE INIT so the msgbuf pipe
cache, the reply bounce pipe, and the fuse_msg_size clamp all see
the raised cap. Root-only knob; failures log at info level (cannot
open sysctl / write failed) and the mount proceeds with the
current kernel value. */
pipe_max_size(0),
readdir("seq"),
rename_exdev(RenameEXDEV::ENUM::PASSTHROUGH),
scheduling_priority(-10),
security_capability(true),
statfs(StatFS::ENUM::BASE),
/* 2026-08-30 pooled bench verdict on tmpfs branches (4 runs x
REPS=9, n=36/cell): seq_write +20.8%, seq_read +52.4%,
small_write +5.0%, small_read +12.1%. 4K cells regress (-16%
write on cold mounts, -4..-12% read). Only the READ side of that
regression is excluded by a size gate: read replies below 128 KiB
use the copy path (fuse_lib_read, fuse.cpp), so 4K reads never hit
splice. Writes have no equivalent gate - fuse_ll_buf_receive_splice
(fuse_lowlevel.cpp) stages every FUSE_WRITE payload into the pipe
regardless of size - so the measured 4K write regression applies
in full under this default. Defaults ON per pre-committed rule
anyway: the small_write/seq_write/seq_read gains outweigh it in
the pooled numbers above, but "excluded by the gates" was true
only for reads. */
splice(true),
statfs_ignore(StatFSIgnore::ENUM::NONE),
symlinkify(false),
symlinkify_timeout(3600),
@@ -174,6 +195,8 @@ Config::Config()
process_thread_queue_depth.ro =
read_thread_count.ro =
scheduling_priority.ro =
pipe_max_size.ro =
splice.ro =
true;
_congestion_threshold.display =
_gid.display =
@@ -273,6 +296,7 @@ Config::Config()
_map["passthrough.max-stack-depth"] = &passthrough_max_stack_depth;
_map["pid"] = &pid;
_map["pin-threads"] = &pin_threads;
_map["pipe-max-size"] = &pipe_max_size;
_map["posix-acl"] = &posix_acl;
_map["process-thread-count"] = &process_thread_count;
_map["process-thread-queue-depth"] = &process_thread_queue_depth;
@@ -284,9 +308,21 @@ Config::Config()
_map["rename-exdev"] = &rename_exdev;
_map["scheduling-priority"] = &scheduling_priority;
_map["security-capability"] = &security_capability;
/* Bare splice-read/write/move (no "no-" prefix) were accepted no-ops
on master (shipped since 2.35.0, see deprecated_options.md) before
ever being restored here. They must stay registered - even though
they predate and are unrelated to the single splice=BOOL knob -
or Config::set() returns -ENOATTR for them (after the
underscore->hyphen normalization below, this also covers the
splice_read/splice_write/splice_move spellings), _process_opt()
treats that as OPT_KEEP, and the token gets forwarded to the
kernel's mount-option parser as unrecognized, failing the mount
with EINVAL - the exact class of regression fixed for "noforget"
elsewhere in this same change. */
_map["splice-move"] = &_dummy;
_map["splice-read"] = &_dummy;
_map["splice-write"] = &_dummy;
_map["splice"] = &splice;
_map["srcmounts"] = &_srcmounts;
_map["statfs"] = &statfs;
_map["statfs-ignore"] = &statfs_ignore;
+16 -1
View File
@@ -156,11 +156,26 @@ public:
ProxyIOPrio proxy_ioprio;
TFSRef<int> read_thread_count;
ConfigU64 readahead;
/* Root-only knob: written to /proc/sys/fs/pipe-max-size during
option cleanup so the vendored libfuse reply/receive pipes can
carry a fuse message larger than the default 1 MiB. Value 0 ==
leave sysctl alone (default); unprivileged writes are silently
dropped so non-root mounts don't spam warnings. Tied to msgbuf
sizing: fuse_msg_size clamp in fuse_init.cpp reads the same
sysctl; receive-pipe capacity and reply bounce-pipe capacity
read it at ensure time. So raising pipe-max-size at mount time
unlocks fuse_msg_size > 1 MiB end-to-end on kernels that allow
the larger messages. */
ConfigU64 pipe_max_size;
FUSE::ReadDir readdir;
RenameEXDEV rename_exdev;
ConfigINT scheduling_priority;
ConfigBOOL security_capability;
StatFS statfs;
/* Master splice switch. The old per-axis splice-read/write/move and
no-splice-read/write/move spellings are accepted as no-ops (see
_dummy) and do not affect this knob. */
ConfigBOOL splice;
StatFSIgnore statfs_ignore;
ConfigBOOL symlinkify;
ConfigS64 symlinkify_timeout;
@@ -170,11 +185,11 @@ private:
TFSRef<int> _congestion_threshold;
Debug _debug;
CfgDummy _dummy;
CfgNoforget _never_forget_nodes;
TFSRef<s64> _gid;
TFSRef<int> _max_background;
TFSRef<fs::path> _mount;
TFSRef<fs::path> _mountpoint;
CfgNoforget _never_forget_nodes;
CfgNoforget _noforget;
CfgNoforget _remember;
CfgNoforget _remember_nodes;
+162
View File
@@ -20,6 +20,7 @@
#include "config.hpp"
#include "fs_readahead.hpp"
#include "pipe_max_size.hpp"
#include "procfs.hpp"
#include "state.hpp"
#include "syslog.hpp"
@@ -33,6 +34,8 @@
#include <fstream>
#include <thread>
#include <unistd.h>
static
void
@@ -77,6 +80,135 @@ _want_if_capable(fuse_conn_info_t *conn_,
#define MAX_FUSE_MSG_SIZE 65535
static const char MAX_PAGES_LIMIT_FILEPATH[] = "/proc/sys/fs/fuse/max_pages_limit";
/* Single place that decides what "splice enabled/disabled" means:
cfg_.splice, all three fuse_cfg.splice_* fields, and the three
FUSE_CAP_SPLICE_* bits in conn_->want always move together. Used
both when the kernel/administrator decision is first applied
(FUSE::init) and when INIT-time pipe negotiation later retracts it
(_negotiate_splice_pipe_capacity below) - a single call site means
a future fourth splice-related flag only needs updating here. */
static
void
_set_splice_enabled(fuse_conn_info_t *conn_,
Config &cfg_,
const bool enabled_)
{
cfg_.splice = enabled_;
fuse_cfg.splice_read = enabled_;
fuse_cfg.splice_write = enabled_;
fuse_cfg.splice_move = enabled_;
if(enabled_)
{
::_want(conn_,FUSE_CAP_SPLICE_READ);
::_want(conn_,FUSE_CAP_SPLICE_WRITE);
::_want(conn_,FUSE_CAP_SPLICE_MOVE);
}
else
{
conn_->want &= ~(FUSE_CAP_SPLICE_READ |
FUSE_CAP_SPLICE_WRITE |
FUSE_CAP_SPLICE_MOVE);
}
}
static
void
_negotiate_splice_pipe_capacity(fuse_conn_info_t *conn_,
Config &cfg_)
{
if(!fuse_cfg.splice_read && !fuse_cfg.splice_write)
return;
/* Splice receive needs one whole request to fit in the per-msgbuf
pipe. The pipe cap is min(g_bufsize, /proc/sys/fs/pipe-max-size),
so max_pages must satisfy
(max_pages + 2) * pagesize <= effective_pipe_max (the +2 covers
the 16B fuse_in_header, the fuse_write_in the receive path
stages before the payload, and the 16B fuse_out_header on
replies).
Modeled on the max_pages_limit write above: when the requested
fuse_msg_size does not fit the current sysctl, attempt to raise
it (root only, best effort, non-fatal), then re-read and clamp
downward if the write didn't take. */
u64 pagesize = (u64)::sysconf(_SC_PAGESIZE);
u32 pipe_max = 1048576;
pipe_max_size::read(pipe_max);
u64 needed = ((u64)cfg_.fuse_msg_size + 2) * pagesize;
if(needed > pipe_max)
{
if(pipe_max_size::write(needed))
{
SysLog::info("{} raised to {} to fit fuse_msg_size",
"/proc/sys/fs/pipe-max-size",
needed);
if(!pipe_max_size::read(pipe_max))
{
SysLog::info("pipe-max-size re-read failed after"
" raise; assuming requested {}",
needed);
/* `needed` is u64; pipe_max is u32. Clamp rather
than truncate - a silent mod-2^32 wrap here
would feed a tiny bogus value into the
fused_pages math just below and could disable
splice entirely (or wrongly clamp fuse_msg_size)
even though the sysctl write above just
succeeded at the real, large value. */
pipe_max = (needed > UINT32_MAX) ? UINT32_MAX : (u32)needed;
}
}
else
{
SysLog::info("splice receive on: unable to raise {}"
" (not root); keeping {}",
"/proc/sys/fs/pipe-max-size",
pipe_max);
}
}
/* fused math: (pipe_max/pages)-2 underflows u32 when pipe_max < 3
pages (reachable via a sysctl set to a tiny value); with exactly
3 pages the clamp yields fused_cap = 1 page - the smallest
workable splice receive. At 2 pages fused_cap would be 0,
negotiating a degenerate max_pages=0 mount, so splice must be
disabled entirely rather than negotiate sizes we can never
honor. */
u64 fused_pages = (pipe_max / pagesize);
if(fused_pages < 3)
{
SysLog::warning("pipe-max-size {} too small for splice"
" (need >= 3 pages); disabling splice"
" read/write/move for this mount",
pipe_max);
/* Runtime disable: fuse_cfg is what the live paths check
(receive re-checks post-INIT, write/reply gates read it
directly). Also flip cfg_.splice (the single mergerfs-level
knob) so the startup Config dump and the splice control-file
xattr report the true, now-disabled state instead of the
stale pre-negotiation value. Also retract the FUSE_CAP_SPLICE_*
bits already OR'd into conn_->want by the caller's earlier
_want() calls (FUSE::init, before _want_if_capable_max_pages
runs) - otherwise do_init's outargflags (fuse_lowlevel.cpp)
would still advertise splice support to the kernel in the
FUSE_INIT reply even though every internal path just got
switched off. */
_set_splice_enabled(conn_,cfg_,false);
return;
}
u32 fused_cap = (u32)(fused_pages - 2);
if(cfg_.fuse_msg_size > fused_cap)
{
SysLog::info("splice receive on: clamping max pages {} -> {}",
(u64)cfg_.fuse_msg_size,
(u64)fused_cap);
cfg_.fuse_msg_size = fused_cap;
}
}
static
void
_want_if_capable_max_pages(fuse_conn_info_t *conn_,
@@ -132,12 +264,18 @@ _want_if_capable_max_pages(fuse_conn_info_t *conn_,
if(::_capable(conn_,FUSE_CAP_MAX_PAGES))
{
::_want(conn_,FUSE_CAP_MAX_PAGES);
_negotiate_splice_pipe_capacity(conn_,cfg_);
fuse_cfg.max_pages = cfg_.fuse_msg_size;
SysLog::info("requesting max pages size of {}",
(u64)cfg_.fuse_msg_size);
}
else
{
SysLog::info("kernel lacks FUSE_MAX_PAGES: capping fuse_msg_size"
" at default {} pages",
(u64)FUSE_DEFAULT_MAX_PAGES_PER_REQ);
cfg_.fuse_msg_size = FUSE_DEFAULT_MAX_PAGES_PER_REQ;
}
}
@@ -207,6 +345,30 @@ FUSE::init(fuse_conn_info_t *conn_)
// ::_want_if_capable(conn_,FUSE_CAP_READDIR_PLUS,&cfg.readdirplus);
::_want_if_capable(conn_,FUSE_CAP_WRITEBACK_CACHE,&cfg.cache_writeback);
::_want_if_capable(conn_,FUSE_CAP_ALLOW_IDMAP,&cfg.allow_idmap);
/* Single splice knob drives all three internal axes: only keep
cfg.splice on when the kernel supports ALL THREE FUSE_CAP_SPLICE_*
bits. Each _capable() check must be independent here - chaining
three calls through the shared ConfigBOOL overload above (as a
prior version of this code did) lets an unsupported flag checked
first latch cfg.splice=false and suppress _want() for a later
flag the kernel DOES support. fuse_cfg.splice_read/write/move are
resynced immediately below (they were previously only set once
from the pre-INIT option value in option_parser.cpp and never
updated here), so the receive/write/reply hot paths - and
_want_if_capable_max_pages just below, which reads them - see the
post-negotiation decision instead of stale unfulfilled intent. */
{
bool splice_capable = (::_capable(conn_,FUSE_CAP_SPLICE_READ) &&
::_capable(conn_,FUSE_CAP_SPLICE_WRITE) &&
::_capable(conn_,FUSE_CAP_SPLICE_MOVE));
if(bool(cfg.splice) && !splice_capable)
SysLog::warning("kernel lacks one or more of"
" FUSE_CAP_SPLICE_READ/WRITE/MOVE: disabling splice"
" for this mount");
_set_splice_enabled(conn_,cfg,bool(cfg.splice) && splice_capable);
}
// ::_want_if_capable(conn_,FUSE_CAP_READDIR_PLUS_AUTO);
::_want_if_capable_max_pages(conn_,cfg);
+68
View File
@@ -0,0 +1,68 @@
/*
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 "ioprio.hpp"
#include "state.hpp"
#include "fuse.h"
#include "fuse_common.h"
/* Return a bufvec referencing the branch fd at the read offset. The
libfuse fuse_lib_read path splices payload directly fd->pipe->/dev/fuse
(kernel-side), avoiding a userspace copy of the payload. Called only
when splice is on and the request size is >= 128KiB (see
fuse_lib_read in fuse.cpp); the payload's memory residency is
irrelevant in that path so we never pread.
*/
int
FUSE::read_buf(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_,
fuse_bufvec *bufp_,
const size_t size_,
const off_t offset_)
{
FileInfo *fi;
/* ioprio is thread-persistent; without SetFrom the branch-file I/O
performed downstream of this handler (the kernel-side splice in
fuse_reply_data_splice_fd runs on this same worker thread) would
inherit whatever priority a previous request left behind. */
ioprio::SetFrom iop(ctx_->pid);
fi = state.get_fi(ctx_,ffi_->fh);
if(not fi)
return -EBADF;
/* ffi_->flags arrives verbatim from the client's FUSE_OPEN and is
passed through to fs::openat, so a client O_DIRECT open yields an
O_DIRECT branch fd. That is still safe here: a splice that
rejects it (EINVAL) or hits EOF (0) returns -1 from
fuse_reply_data_splice_fd and fuse_lib_read_fd_splice falls back
to the plain pread read path. */
*bufp_ = fuse_bufvec_fd(fi->fd,offset_,size_,
(fuse_buf_flags)(FUSE_BUF_IS_FD|
FUSE_BUF_FD_SEEK|
FUSE_BUF_FD_RETRY));
return 0;
}
+32
View File
@@ -0,0 +1,32 @@
/*
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.h"
#include "fuse_common.h"
namespace FUSE
{
int read_buf(const fuse_req_ctx_t *ctx,
const fuse_file_info_t *ffi,
fuse_bufvec *bufp,
const size_t size,
const off_t offset);
}
+234
View File
@@ -0,0 +1,234 @@
/*
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_write.hpp"
#include "fuse_write_buf.hpp"
#include "config.hpp"
#include "errno.hpp"
#include "fileinfo.hpp"
#include "fs_getfl.hpp"
#include "ioprio.hpp"
#include "state.hpp"
#include "fuse.h"
#include <cstring>
#include <fcntl.h>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <sys/ioctl.h>
/* Pure splice copy: pipe -> branch fd. Safe to run under fi->mutex's
shared lock. Returns fuse_buf_copy's result: >= 0 bytes moved (the
srcv_ idx/off mark what was consumed), or -errno with NOTHING moved
(the source pipe is left intact). Never delegates and never locks
beyond the caller's shared lock. */
static
int
_write_buf_splice(const int fd_,
fuse_bufvec *srcv_,
const off_t offset_,
const size_t size_)
{
fuse_bufvec dst = fuse_bufvec_fd(fd_,offset_,size_,
(fuse_buf_flags)(FUSE_BUF_IS_FD|
FUSE_BUF_FD_SEEK));
const fuse_buf_copy_flags cpflags =
(fuse_buf_copy_flags)(FUSE_BUF_SPLICE_MOVE | FUSE_BUF_SPLICE_NONBLOCK);
ssize_t copied = fuse_buf_copy(&dst,srcv_,cpflags);
return (int)copied;
}
/* Mem fallback for whatever the splice could not move (partial splice,
O_DIRECT branch fd rejecting splice, or a spliced prefix followed by
an error): read the remaining pipe payload into a heap buffer and
delegate to FUSE::write - lock-free by design, because FUSE::write
takes fi->mutex itself and owns the full move-on-enospc handling
(re-check, move, pwriten retry). Any duplication of that logic here
would re-enter the same non-recursive mutex and self-deadlock. */
static
int
_write_buf_fallback(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_,
fuse_bufvec *srcv_,
const off_t offset_)
{
size_t remaining = ((size_t)srcv_->buf[srcv_->idx].size - srcv_->off);
if(remaining == 0)
return 0;
/* A failed splice attempt may have already consumed pipe bytes
without advancing srcv_: on EINVAL (e.g. O_APPEND branch fds
reject splice with an explicit offset) fuse_buf_splice falls
back to fuse_buf_fd_to_fd, which reads from the pipe into a
stack buffer and can fail (or return short) after that read,
dropping the bytes it consumed. Reading `remaining` would then
block forever on a pipe whose write end this request itself
holds. FIONREAD bounds the read; a short pipe means payload
bytes are gone, so fail the request - the app retries, and
msgbuf_free's drain cleans the pipe. */
int avail;
if(ioctl(srcv_->buf[srcv_->idx].fd,FIONREAD,&avail) == -1)
return -errno;
/* avail is `int`; guard the sign before the size_t comparison below
widens a negative value into a huge one and defeats this exact
short-pipe check (avail should never be negative in practice, but
the cast must not silently trust that). */
if((avail < 0) || ((size_t)avail < remaining))
return -EIO;
std::unique_ptr<char[]> buf(new(std::nothrow) char[remaining]);
if(!buf)
return -ENOMEM;
size_t to_read = remaining;
while(to_read > 0)
{
ssize_t nr;
nr = ::read(srcv_->buf[srcv_->idx].fd,buf.get() + (remaining - to_read),to_read);
if(nr == -1)
{
if(errno == EINTR)
continue;
return -errno;
}
if(nr == 0)
return -EIO;
to_read -= nr;
}
return FUSE::write(ctx_,ffi_,buf.get(),remaining,offset_);
}
int
FUSE::write_buf(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_,
struct fuse_bufvec *srcv_,
off_t offset_)
{
ioprio::SetFrom iop(ctx_->pid);
FileInfo *fi = state.get_fi(ctx_,ffi_->fh);
if(fi == nullptr)
return -EBADF;
/* O_APPEND branch fds deterministically reject splice(2) with
EINVAL (splice requires an explicit, seekable destination offset;
O_APPEND ignores any offset and always writes at EOF instead), so
_write_buf_splice would immediately fall through fuse_buf_splice
into fuse_buf_fd_to_fd's byte-at-a-time bounce copy for the WHOLE
payload. That bounce loop reads a chunk from the (unrereadable,
one-shot) pipe before writing it to the branch fd; if the write
fails partway (e.g. ENOSPC on a full branch after earlier chunks
already landed), the just-read chunk is silently lost with no way
to recover it, and _write_buf_fallback below can only refuse the
request outright (its FIONREAD check trips) rather than let
FUSE::write relocate it to another branch, because the payload it
would need to resend is already gone. Skip the splice attempt
entirely for O_APPEND: go straight to the fallback here, while
the full unread payload is still intact in the pipe, so the
lossless heap-buffer path (and FUSE::write's moveonenospc retry)
is used from the start instead of being reached only after damage
is done. cache.writeback=true strips O_APPEND from the branch
fd itself (fuse_open.cpp: _tweak_flags_cache_writeback, gated
on cfg.cache_writeback - an option independent of cache.files),
so this only fires when cache.writeback is left at its default
of false, regardless of the cache.files setting.
O_DIRECT branch fds (ffi_->flags forwards the client's O_DIRECT
open verbatim - see fuse_read_buf.cpp) can likewise reject
splice(2) with EINVAL. Unlike O_APPEND that alone would just
route the whole payload through the same lossless
fuse_buf_fd_to_fd bounce (no data-loss hazard, since the offset
is still explicit/seekable) - but a splice that partially
succeeds before an O_DIRECT-related short count would leave
_write_buf_fallback resuming at a non-block-aligned offset/size,
which O_DIRECT destinations can reject even though the original,
client-aligned request would have succeeded. Bypass splice
entirely for O_DIRECT too so the fallback always sees the whole,
originally-aligned request. */
if(ffi_->flags & (O_APPEND|O_DIRECT))
return _write_buf_fallback(ctx_,ffi_,srcv_,offset_);
/* The check above reads fuse_write_in.flags, which the kernel only
fills in for writes issued from a kiocb. Writeback flushes go
through fuse_writepage_args_setup -> fuse_write_args_fill, which
never assigns it, so a FUSE_WRITE_CACHE request (ffi_->writepage)
always arrives with flags == 0 and slips past the guard into the
lossy bounce described above. The branch fd really can be
O_APPEND/O_DIRECT there: with cache.writeback=false via a
MAP_SHARED mmap of a file opened O_APPEND (the O_APPEND strip in
fuse_open.cpp is gated on cfg.cache_writeback), and with
cache.writeback=true via an O_DIRECT handle, since writepages
picks an arbitrary write-capable fh rather than the one that
dirtied the pages. Ask the fd itself instead of trusting the
request, and treat an unreadable answer as the worst case - the
fallback is always correct, only slower. */
if(ffi_->writepage)
{
int fl = fs::getfl(fi->fd);
if((fl < 0) || (fl & (O_APPEND|O_DIRECT)))
return _write_buf_fallback(ctx_,ffi_,srcv_,offset_);
}
/* Splice under the shared lock only - it makes no nested calls.
Mirrors _write's discipline (fuse_write.cpp) so parallel direct
writes serialize against mem-path writers on the same file. */
size_t size = fuse_buf_size(srcv_);
ssize_t copied;
{
std::shared_lock<std::shared_mutex> slk(fi->mutex);
copied = _write_buf_splice(fi->fd,srcv_,offset_,size);
}
/* copied < 0: nothing moved, pipe intact (fuse_buf_copy only
returns errors when zero bytes were consumed).
srcv_->idx < srcv_->count: partial move, remainder still queued.
Either way the fallback drains the pipe and delegates to the
mem write path lock-free; FUSE::write owns ENOSPC/move/retry. */
if(copied < 0 || srcv_->idx < srcv_->count)
{
off_t done = (copied > 0) ? (off_t)copied : (off_t)0;
int rv = _write_buf_fallback(ctx_,ffi_,srcv_,offset_ + done);
if(rv < 0)
{
/* `done` bytes are already durably written to the branch
fd via the splice above; a request-level error here must
not discard that progress. Per this codebase's own
write() contract (fuse_write.cpp: "N bytes written
(short writes included)"), report the short write
instead of a bare failure so the kernel/application
don't believe zero bytes landed when `done` actually
did. */
return (done > 0) ? (int)done : rv;
}
return (int)(done + rv);
}
return (int)copied;
}
+31
View File
@@ -0,0 +1,31 @@
/*
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.h"
namespace FUSE
{
int
write_buf(const fuse_req_ctx_t *ctx,
const fuse_file_info_t *ffi,
struct fuse_bufvec *srcv,
off_t offset);
}
+16
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"
@@ -83,6 +84,7 @@
#include "fuse_unlink.hpp"
#include "fuse_utimens.hpp"
#include "fuse_write.hpp"
#include "fuse_write_buf.hpp"
#include "fuse.h"
@@ -147,6 +149,20 @@ _get_fuse_operations(struct fuse_operations &ops_,
ops_.unlink = FUSE::unlink;
ops_.utimens = FUSE::utimens;
ops_.write = (nullrw_ ? FUSE::write_null : FUSE::write);
/* write_buf is held NULL under nullrw: FUSE::write_null discards the
buffer without reading it. NOTE the receive path still stages the
payload in the pipe when splice is on (staging is gated on
fuse_cfg.splice_write, not on ops registration) - write_null
never dereferences the data pointer, and msgbuf_free's bounded
drain discards the leftover pipe bytes. Any future nullrw write
implementation that reads the buffer must handle payload-in-pipe. */
ops_.write_buf = ((nullrw_ || !cfg.splice) ? nullptr : FUSE::write_buf);
/* read_buf is held NULL under nullrw (read_null already skips all
I/O - nothing on the branch to splice from). When splice is on,
read_buf returns a bufvec referencing the branch fd and the
kernel splices the payload straight from the branch into /dev/fuse
via a pipe. */
ops_.read_buf = ((nullrw_ || !cfg.splice) ? nullptr : FUSE::read_buf);
return;
}
+22
View File
@@ -17,6 +17,7 @@
#define FMT_HEADER_ONLY
#include "config.hpp"
#include "fuse_cfg.hpp"
#include "ef.hpp"
#include "errno.hpp"
#include "fmt/core.h"
@@ -25,6 +26,7 @@
#include "fs_statvfs_cache.hpp"
#include "hw_cpu.hpp"
#include "num.hpp"
#include "pipe_max_size.hpp"
#include "policy.hpp"
#include "str.hpp"
#include "syslog.hpp"
@@ -290,6 +292,26 @@ _cleanup_options()
{
if(!cfg.symlinkify)
cfg.symlinkify_timeout = -1;
/* splice backing lives in fuse_cfg so fuse_lowlevel_new_common can
select the receive path before FUSE::init ever runs. fuse_init only
flips bits off if the kernel didn't advertise the capability. The
single cfg.splice knob drives all three axes. */
fuse_cfg.splice_move = bool(cfg.splice);
fuse_cfg.splice_read = bool(cfg.splice);
fuse_cfg.splice_write = bool(cfg.splice);
/* pipe-max-size: early sysctl write for root-run mergerfs so the
reply bounce pipe and msgbuf receive pipes see the larger capacity
when they are first created (and the fuse_msg_size clamp in
fuse_init.cpp reads the raised value when negotiating INIT). */
if((u64)cfg.pipe_max_size != 0)
{
if(!pipe_max_size::write((u64)cfg.pipe_max_size))
SysLog::info("pipe-max-size: unable to raise sysctl to {}"
" (not root, or write failed); keeping kernel default",
(u64)cfg.pipe_max_size);
}
}
namespace options
+78
View File
@@ -0,0 +1,78 @@
/*
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 "pipe_max_size.hpp"
#include "scope_guard/scope_guard.hpp"
#include <array>
#include <charconv>
#include <fcntl.h>
#include <unistd.h>
static const char PIPE_MAX_SIZE_FILEPATH[] = "/proc/sys/fs/pipe-max-size";
bool
pipe_max_size::read(u32 &out_)
{
int fd = ::open(PIPE_MAX_SIZE_FILEPATH,O_RDONLY|O_CLOEXEC);
if(fd == -1)
return false;
DEFER { ::close(fd); };
std::array<char,16> buf;
ssize_t nr = ::read(fd,buf.data(),buf.size());
if(nr <= 0)
return false;
unsigned long v;
auto [end,ec] = std::from_chars(buf.data(),buf.data() + nr,v);
if(ec != std::errc())
return false;
/* Clamp rather than treat as a read failure: a legitimately
huge sysctl (e.g. an admin-set multi-GiB pipe-max-size,
written verbatim as a u64 by the pipe-max-size= option)
must not be mistaken for "couldn't read it" and fall back
to a small default in the caller - that fallback can then
silently shrink the admin's explicit setting when the
caller reacts to a seemingly-tiny sysctl. */
out_ = (v > UINT32_MAX) ? UINT32_MAX : (u32)v;
return true;
}
bool
pipe_max_size::write(cu64 v_)
{
int fd = ::open(PIPE_MAX_SIZE_FILEPATH,O_WRONLY|O_CLOEXEC);
if(fd == -1)
return false;
DEFER { ::close(fd); };
std::array<char,32> buf;
auto [end,ec] = std::to_chars(buf.data(),buf.data() + buf.size(),(u64)v_);
if(ec != std::errc())
return false;
ssize_t len = (end - buf.data());
ssize_t wr = ::write(fd,buf.data(),len);
return (wr == len);
}
+27
View File
@@ -0,0 +1,27 @@
/*
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 "base_types.h"
namespace pipe_max_size
{
bool read(u32 &out_);
bool write(cu64 v_);
}
+415
View File
@@ -0,0 +1,415 @@
#!/bin/bash
# bench-splice.sh -- A/B benchmark of mergerfs splice
# against the mergefs OFF baseline and a pristine-HEAD baseline binary.
#
# Run: bash tools/bench-splice.sh [REPS]
# Env: BASE_REPO (path to pristine HEAD build dir), SIZE_MIB, REPS
# note: requires `unshare -Urm` permissions for an unprivileged FUSE mount.
# Resolve a working `stat -c %s` once: this host's unshare PATH can
# front a non-GNU stat (aurora) that rejects -c.
STAT_BIN="$(command -v stat)"
if ! "${STAT_BIN}" -c %s /dev/null >/dev/null 2>&1; then
STAT_BIN=/usr/bin/stat
fi
set -euo pipefail
export LC_ALL=C # dd prints locale-formatted numbers; parse_mibs needs C
REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
BIN="${REPO}/build/mergerfs"
BASE_REPO="${BASE_REPO:-/tmp/mfs-base}"
BASE_BIN="${BASE_REPO}/build/mergerfs"
B=/dev/shm/bench-splice
REPS="${REPS:-${1:-5}}"
# --- OOM guard: pick SIZE_MIB so mergefs-blob fits comfortably into MemAvailable
mem_available_kib() { awk '/^MemAvailable:/{print $2}' /proc/meminfo; }
pick_size_mib() {
local want=(256 128 64)
local w
for w in "${want[@]}"; do
local need=$(( w * 2 * 1024 )) # blob on branch + mergefs page cache
local have; have=$(mem_available_kib)
if (( have >= need )); then echo "$w"; return; fi
done
echo "INSUFFICIENT-MEMORY" >&2
return 1
}
SIZE_MIB="${SIZE_MIB:-$(pick_size_mib)}"
# --- helpers -----------------------------------------------------------
die() { echo "bench-splice: $*" >&2; exit 1; }
need_binary() { [[ -x "$1" ]] || die "missing binary: $1"; }
need_splice_binary() {
# --version comes from src/version.hpp, which incremental makes do not
# refresh; assert the binary actually contains the code under test.
# NB: consume strings fully - grep -q closes the pipe early, and with
# pipefail the resulting SIGPIPE would fail the check spuriously.
local marker="$2"
local out
out=$(strings "$1" | grep "${marker}" || true)
[[ -n "${out}" ]] \
|| die "binary $1 lacks marker '${marker}' (stale build? run make)"
}
ensure_unshared() {
# Re-exec under unshare -Urm if we're not already root in a user ns
# (mount requires CAP_SYS_ADMIN in this host's delegate).
if (( EUID != 0 )) && [[ -z "${BENCH_INNS:-}" ]]; then
BENCH_INNS=1 exec unshare -Urm -- "$0" "$@"
fi
}
# --- mergefs process management -----------------------------------------
declare -g MNT="" BR1="" BR2="" MPID=""
mfs_stop() {
if [[ -n "${MPID}" ]]; then
kill "${MPID}" 2>/dev/null || true
wait "${MPID}" 2>/dev/null || true
MPID=
fi
pkill -9 -f "$BIN .*${B}/mnt" 2>/dev/null || true
pkill -9 -f "$BASE_BIN .*${B}/mnt" 2>/dev/null || true
if [[ -n "${MNT}" ]] && grep -qs "${B}/mnt" /proc/mounts; then
fusermount3 -u "${MNT}" >/dev/null 2>&1 || umount "${MNT}" 2>/dev/null || true
fi
}
cleanup() {
mfs_stop
payload_src_fini
rm -rf "${B}"
}
trap cleanup EXIT
trap 'cleanup; exit 1' INT TERM
mfs_start() {
local binary="$1"; shift
local opts="$1"; shift
BR1="${B}/b1"; BR2="${B}/b2"; MNT="${B}/mnt"
mkdir -p "${BR1}" "${BR2}" "${MNT}"
"${binary}" -f "${BR1}:${BR2}" "${MNT}" -o "${opts}" >/dev/null 2>&1 &
MPID=$!
# wait for mount
local t=0
while ! grep -qs " ${MNT} " /proc/mounts; do
sleep 0.2
t=$(( t + 1 ))
if (( t > 50 )); then
die "mount did not come up for ${binary}"
fi
done
sleep 0.5 # let it warm INIT
}
# --- one experiment ------------------------------------------------------
# run_4phases <binary> <mount-opts> <label>
# The four phases: seq_write, seq_read, small_write, small_read
# Writes results rows "label test rep mibs" to $RESULTS.
RESULTS=${RESULTS:-/tmp/bench-splice.$(date +%Y%m%d%H%M%S).results}
: > "${RESULTS}"
parse_mibs() {
# dd's final stderr line: 'NNN bytes (X biB) copied, T s, R MB/s' where its
# "MB/s" is decimal. Recompute MiB/s exactly from bytes and seconds.
tail -n 1 "$1" | awk '
{ for (i = 1; i <= NF; i++) {
if ($i == "bytes") bytes = $(i-1)
if ($i == "copied,") secs = $(i+1)
}
if (bytes && secs > 0) printf "%.3f", bytes / 1048576.0 / secs
}'
}
phase_run() {
local label="$1" test="$2" rep="$3"; shift 3
local lav tmpf rc
tmpf=$(mktemp)
# Explicitly capture dd's exit status: a `;`-joined
# "dd ...; parse_mibs" (the previous form) ran parse_mibs
# unconditionally, so a dd that failed partway (e.g. EIO from a
# splice fallback) could still leave a plausible-looking partial
# "N bytes ... copied" summary line in ${tmpf} for parse_mibs to
# extract, recording a corrupt-transport measurement as if it were
# a valid one instead of aborting the run.
rc=0
"$@" 2>"${tmpf}" >/dev/null || rc=$?
if [[ ${rc} -ne 0 ]]; then
cat "${tmpf}" >&2
rm -f "${tmpf}"
die "dd failed (exit ${rc}) for $label/$test/rep$rep"
fi
lav="$(parse_mibs "${tmpf}" || true)"
rm -f "${tmpf}"
[[ -n "${lav}" ]] || die "failed to parse MiB/s from dd ($label/$test/rep$rep)"
echo "$label $test $rep $lav" >> "${RESULTS}"
}
wipe_targets() {
rm -f "${MNT}/blob" "${BR1}/blob" "${BR2}/blob"
sync
}
check_integrity() {
# verify the file on disk matches what's seen through the mount, and
# that both carry the full expected size (agreement on a truncated
# blob is NOT integrity).
# (set -e + pipefail safe: substitutions must never abort the script)
local expected="$1"
local mhash bhash msize bsize b_path
mhash=$(sha256sum "${MNT}/blob" 2>/dev/null | awk '{print $1}') || true
msize=$("${STAT_BIN}" -c %s "${MNT}/blob" 2>/dev/null) || msize=""
bhash=""
bsize=""
b_path=""
if [[ -f "${BR1}/blob" ]]; then b_path="${BR1}/blob"; fi
if [[ -z "${b_path}" && -f "${BR2}/blob" ]]; then b_path="${BR2}/blob"; fi
if [[ -n "${b_path}" ]]; then
bhash=$(sha256sum "${b_path}" 2>/dev/null | awk '{print $1}') || true
bsize=$("${STAT_BIN}" -c %s "${b_path}" 2>/dev/null) || bsize=""
fi
[[ -n "${bhash}" ]] || die "blob not found on any branch (integrity)"
if [[ "${bsize}" != "${expected}" || "${msize}" != "${expected}" ]]; then
die "blob size ${bsize}/${msize} != expected ${expected} (truncated write, integrity)"
fi
if [[ "${mhash}" != "${bhash}" ]]; then
die "sha256 mismatch branch ${b_path} vs mount (integrity check)"
fi
}
drop_caches() {
# Best effort - and normally a no-op here: vm sysctls are not
# namespaced, so writing drop_caches from inside unshare -Urm
# requires CAP_SYS_ADMIN in the INITIAL userns and fails with EPERM
# (masked below). Read phases on tmpfs branches are therefore
# cache-warm; they measure the transport of cached pages, not cold
# disk. sync still runs.
sync
echo 3 > /proc/sys/vm/drop_caches 2>/dev/null || true
}
# Non-zero payload source: the integrity gate must be able to see
# stale-pool-memory corruption (correct size, wrong bytes). An
# all-zero payload makes stale pool contents indistinguishable from
# the requested data, so the sha256 gate passes green over corrupted
# transport. One urandom file per bench run, reused across phases.
PAYLOAD_SRC=""
payload_src_init() {
# Sized to the LARGEST payload any arm writes: SIZE_MIB for the
# main arms, and SIZE_MIB_4K MiB for the 4K arm (RUN_4K defaults
# on and its size is independent of SIZE_MIB). A short pattern
# file would make the write dd run short and check_integrity die
# 'truncated write'. Lives in /dev/shm (same as B) so the read
# source costs memory speed, not disk.
local need
need=$(( SIZE_MIB > ${SIZE_MIB_4K:-64} ? SIZE_MIB : ${SIZE_MIB_4K:-64} ))
PAYLOAD_SRC=$(mktemp /dev/shm/bench-splice-payload.XXXXXX 2>/dev/null || mktemp /tmp/bench-splice-payload.XXXXXX)
dd if=/dev/urandom of="${PAYLOAD_SRC}" bs=1M count="${need}" status=none
}
payload_src_fini() {
# Under `set -e`, "[[ cond ]] && cmd" as a function's only statement
# returns the (nonzero) test result when cond is false - e.g. when
# this runs from the EXIT trap before payload_src_init ever set
# PAYLOAD_SRC (an early die() for a missing/stale binary). That
# would make cleanup() exit right here and skip its own later
# `rm -rf "${B}"`, leaking the whole scratch tree. Always return 0.
[[ -n "${PAYLOAD_SRC}" ]] && rm -f "${PAYLOAD_SRC}"
return 0
}
# Write (+integrity check) then drop-caches+read one (name,bs,count) pair
# against ${MNT}/blob. Shared by arm_one_rep (seq/small, bs=1M/128K) and
# arm_4k_rep (tiny, bs=4K). NOT used by arm_native: that arm targets the
# raw branch with no mergefs running, so it has nothing to check
# integrity against and no page cache pass-through worth dropping (see
# its own comment) - forcing it through this helper would need those
# checks toggled off by flag, trading a clear structural difference for
# a harder-to-audit boolean.
run_pair() {
local label="$1" name="$2" bs_arg="$3" count="$4" bytes_per_unit="$5" rep="$6"
local mnt="${MNT}"
wipe_targets
phase_run "${label}" "${name}_write" "${rep}" \
dd if="${PAYLOAD_SRC}" of="${mnt}/blob" bs="${bs_arg}" count="${count}" conv=fdatasync oflag=nocache
check_integrity $(( count * bytes_per_unit ))
drop_caches
phase_run "${label}" "${name}_read" "${rep}" \
dd if="${mnt}/blob" of=/dev/null bs="${bs_arg}" iflag=nocache
}
arm_one_rep() {
local label="$1" rep="$2"
run_pair "${label}" seq 1M "${SIZE_MIB}" 1048576 "${rep}"
run_pair "${label}" small 128K "$(( SIZE_MIB * 8 ))" 131072 "${rep}"
rm -f "${MNT}/blob"
sync
}
# Same arm at bs=4K. Uses SIZE_MIB_4K (default 64) so iteration count
# doesn't run 10x the main arm for equal signal. Writes are capped
# proportionally to SIZE_MIB_4K, reads also capped.
arm_4k_rep() {
local label="$1" rep="$2"
local sz="${SIZE_MIB_4K:-64}"
run_pair "${label}_4k" tiny 4K "$(( sz * 256 ))" 4096 "${rep}"
rm -f "${MNT}/blob"
sync
}
# native baseline: same 4 dd patterns against raw branch directly, once
arm_native() {
local b="${BR1}"
[[ -d "${b}" ]] || die "no branch dir"
phase_run native seq_write 1 \
dd if="${PAYLOAD_SRC}" of="${b}/blob" bs=1M count="${SIZE_MIB}" conv=fdatasync oflag=nocache
phase_run native seq_read 1 \
dd if="${b}/blob" of=/dev/null bs=1M iflag=nocache
rm -f "${b}/blob"; sync
phase_run native small_write 1 \
dd if="${PAYLOAD_SRC}" of="${b}/blob" bs=128K count=$(( SIZE_MIB * 8 )) conv=fdatasync oflag=nocache
phase_run native small_read 1 \
dd if="${b}/blob" of=/dev/null bs=128K iflag=nocache
rm -f "${b}/blob"; sync
}
# --- main ---------------------------------------------------------------
ensure_unshared "$@"
need_binary "${BIN}"
need_binary "${BASE_BIN}"
need_splice_binary "${BIN}" "reply-fd-splice"
mkdir -p "${B}" "${B}/b1" "${B}/b2" "${B}/mnt"
payload_src_init
# header
{
echo "== bench-splice =="
echo "kernel: $(uname -r)"
echo "binary: $(${BIN} --version 2>/dev/null | head -1 || echo mergerfs)"
echo "baseline_binary: $(${BASE_BIN} --version 2>/dev/null | head -1 || echo mergerfs)"
echo "nproc: $(nproc)"
echo "mem_available_kib: $(mem_available_kib)"
echo "size_mib: ${SIZE_MIB}"
echo "reps: ${REPS}"
echo
} | tee /dev/stderr
# native first (no mergefs running); BR1/BR2/MNT point at bench branches
# before any mergefs is started so arm_native can dd directly.
BR1="${B}/b1"; BR2="${B}/b2"; MNT="${B}/mnt"
arm_native
# baseline binary (HEAD): splice options are meaningless to it
mfs_start "${BASE_BIN}" "defaults,minfreespace=0"
for rep in $(seq 1 "${REPS}"); do
arm_one_rep baseline "${rep}"
done
mfs_stop
# interleave off/on arms
for rep in $(seq 1 "${REPS}"); do
mfs_start "${BIN}" "minfreespace=0,splice=false"
arm_one_rep off "${rep}"
mfs_stop
# splice=on (current tree, options ON)
mfs_start "${BIN}" "minfreespace=0,splice=true"
arm_one_rep on "${rep}"
mfs_stop
done
# optional 4K side arm, separate label so rows never aggregate with the
# main on/off pools. Defaults: RUN_4K=1 REPS_4K=3 SIZE_MIB_4K=64.
if [[ "${RUN_4K:-1}" == "1" ]]; then
for rep in $(seq 1 "${REPS_4K:-3}"); do
mfs_start "${BIN}" "minfreespace=0,splice=false"
arm_4k_rep off "${rep}"
mfs_stop
mfs_start "${BIN}" "minfreespace=0,splice=true"
arm_4k_rep on "${rep}"
mfs_stop
done
fi
echo
echo "== raw results (label test rep MiB/s) =="
cat "${RESULTS}"
echo
# min/median/max per label x test
echo "== min / median / max per (label,test) =="
awk '
{ x[$1, $2] = x[$1, $2] " " $4 }
END {
for (k in x) {
split(k, kk, SUBSEP)
n = split(x[k], v, " ")
# bubble sort small
for (i = 1; i <= n; i++)
for (j = i+1; j <= n; j++)
if (v[i]+0 > v[j]+0) { t=v[i]; v[i]=v[j]; v[j]=t }
mn = v[1]; mx = v[n]
if (n % 2 == 1) md = v[(n+1)/2]
else md = (v[n/2]+0 + v[n/2+1]+0) / 2
printf "%-10s %-12s n=%d min=%0.1f med=%0.1f max=%0.1f\n", kk[1], kk[2], n, mn, md, mx
}
}' "${RESULTS}" | sort -k1,1 -k2,2
echo
echo "== delta table (median of PAIRED per-rep deltas: on vs off, on vs baseline) =="
python3 - "${RESULTS}" <<'PY'
import sys, statistics
from collections import defaultdict
rows=[]
with open(sys.argv[1]) as f:
for line in f:
label,test,rep,val=line.split()
rows.append((label,test,int(rep),float(val)))
# (label,test) -> {rep: val}
by_rep=defaultdict(dict)
for label,test,rep,val in rows:
by_rep[(label,test)][rep]=val
def med(l,t):
d=by_rep.get((l,t))
return statistics.median(d.values()) if d else None
# paired per-rep deltas: join two labels by rep number, take the
# difference only where BOTH arms have that rep. The bench interleaves
# off/on per rep precisely so slow drift in host load cancels within a
# pair; pooled medians discard that pairing and were measured 20%+ off
# on this shared host for identical code.
def paired_deltas(a,b,t):
da=by_rep.get((a,t),{}); db=by_rep.get((b,t),{})
common=sorted(set(da)&set(db))
return [da[r]-db[r] for r in common], len(common)
tests=["seq_write","seq_read","small_write","small_read"]
print(f"{'test':<12} {'native':>10} {'baseline':>10} {'off':>10} {'on':>10} {'on/off':>7} {'on/base':>7} pairs")
for t in tests:
n=med("native",t); b=med("baseline",t); o=med("off",t); on=med("on",t)
d_on_off,no_=paired_deltas("on","off",t)
d_on_base,nb=paired_deltas("on","baseline",t)
def pctp(d,base):
# per-column guard: on/base must not depend on the off arm
if not d or not base: return " n/a "
return f"{100.0*statistics.median(d)/base:+.1f}%"
p1=pctp(d_on_off,o)
p2=pctp(d_on_base,b)
print(f"{t:<12} {n if n else 0:>10.1f} {b if b else 0:>10.1f} {o if o else 0:>10.1f} {on if on else 0:>10.1f} {p1:>7} {p2:>7} {no_}/{nb}")
PY
+9
View File
@@ -201,11 +201,20 @@ struct fuse_operations
const char *data,
size_t size,
off_t off);
int (*write_buf)(const fuse_req_ctx_t *,
const fuse_file_info_t *,
struct fuse_bufvec *,
off_t);
int (*read)(const fuse_req_ctx_t *,
const fuse_file_info_t *ffi,
char *buf,
size_t size,
off_t off);
int (*read_buf)(const fuse_req_ctx_t *,
const fuse_file_info_t *,
struct fuse_bufvec *bufp,
size_t size,
off_t off);
int (*fallocate)(const fuse_req_ctx_t *,
const uint64_t,
int,
+10
View File
@@ -49,6 +49,16 @@ public:
int process_thread_queue_depth = 2;
std::string pin_threads = "false";
/* 2026-08-29 fd-splice benchmark: seq_write +8%..+16%, seq_read
+19%..+46%, small_write ~0%/+6%, small_read +15%..+17% in both
runs -> Step 8 rule satisfied, defaults ON.
Prior mem-bounce baseline (vmsplice payload, no read_buf fd
reply) had small_read at -22%, so this path depends on the fd-
referenced read_buf being wired. */
bool splice_read = true;
bool splice_write = true;
bool splice_move = true;
u16 request_timeout = 0;
};
+87
View File
@@ -85,6 +85,9 @@ struct fuse_file_info_t
* FUSE_CAP_EXPORT_SUPPORT: filesystem handles lookups of "." and ".."
* FUSE_CAP_BIG_WRITES: filesystem can handle write size larger than 4kB
* FUSE_CAP_DONT_MASK: don't apply umask to file mode on create operations
* FUSE_CAP_SPLICE_WRITE: ability to use splice() to write to the fuse device
* FUSE_CAP_SPLICE_MOVE: ability to move data to the fuse device with splice()
* FUSE_CAP_SPLICE_READ: ability to use splice() to read from the fuse device
* FUSE_CAP_IOCTL_DIR: ioctl support on directories
* FUSE_CAP_CACHE_SYMLINKS: cache READLINK responses
*/
@@ -94,6 +97,11 @@ struct fuse_file_info_t
#define FUSE_CAP_EXPORT_SUPPORT (1ULL << 4)
#define FUSE_CAP_BIG_WRITES (1ULL << 5)
#define FUSE_CAP_DONT_MASK (1ULL << 6)
/* Same values as the FUSE_SPLICE_WRITE/MOVE/READ wire
flags in fuse_kernel.h */
#define FUSE_CAP_SPLICE_WRITE (1ULL << 7)
#define FUSE_CAP_SPLICE_MOVE (1ULL << 8)
#define FUSE_CAP_SPLICE_READ (1ULL << 9)
#define FUSE_CAP_FLOCK_LOCKS (1ULL << 10)
#define FUSE_CAP_IOCTL_DIR (1ULL << 11)
#define FUSE_CAP_READDIR_PLUS (1ULL << 13)
@@ -315,6 +323,61 @@ struct fuse_buf {
off_t pos;
};
/* Gather buffer declarations */
#define FUSE_BUFVEC_INIT(size__) \
/* cppcheck-suppress unassignedVariable */ \
(struct fuse_bufvec) { \
/* .count= */ 1, \
/* .idx = */ 0, \
/* .off = */ 0, \
{ /* .size = */ (size__), /* .flags = */ (enum fuse_buf_flags) 0, \
/* .mem = */ nullptr, /* .fd = */ -1, /* .pos = */ 0, } \
}
struct fuse_bufvec {
/**
* Number of buffers in the array
*/
size_t count;
/**
* Current buffer index
*/
size_t idx;
/**
* Current offset in the buffer
*/
size_t off;
/**
* Array of buffers
*/
struct fuse_buf buf[1];
};
/**
* Get the total size of data in a fuse buffer vector
*
* @param bufv buffer vector
* @return computed number of bytes
*/
size_t fuse_buf_size(const struct fuse_bufvec *bufv);
/**
* Copy data from one buffer vector to another
*
* This is mostly useful for a filesystem that uses
* buffers to refer to the data being written or read
*
* @param dstv destination buffer vector
* @param srcv source buffer vector
* @param flags flags passed to the buffer copy
* @return actual number of bytes copied or -errno on error
*/
ssize_t fuse_buf_copy(struct fuse_bufvec *dstv, struct fuse_bufvec *srcv,
enum fuse_buf_copy_flags flags);
/* ----------------------------------------------------------- *
* Signal handling *
* ----------------------------------------------------------- */
@@ -341,3 +404,27 @@ int fuse_set_signal_handlers(struct fuse_session *se);
void fuse_remove_signal_handlers(struct fuse_session *se);
EXTERN_C_END
/* C++-only convenience, deliberately declared OUTSIDE EXTERN_C_BEGIN/
EXTERN_C_END above: this is not part of the C-linkage libfuse-
compatible API, just a small typed factory used by the splice
read/write handlers (src/fuse_read_buf.cpp, src/fuse_write_buf.cpp,
vendored/libfuse/lib/fuse.cpp's fuse_lib_write) so they build a
single-entry, fd-referencing fuse_bufvec through one call instead of
each hand-assigning buf[0].flags/fd/pos individually. */
static
inline
struct fuse_bufvec
fuse_bufvec_fd(const int fd_,
const off_t pos_,
const size_t size_,
const fuse_buf_flags flags_)
{
struct fuse_bufvec bufv = FUSE_BUFVEC_INIT(size_);
bufv.buf[0].flags = flags_;
bufv.buf[0].fd = fd_;
bufv.buf[0].pos = pos_;
return bufv;
}
+5
View File
@@ -284,6 +284,11 @@ int fuse_reply_copy_file_range_64(fuse_req_t *req, size_t count);
*/
int fuse_reply_buf(fuse_req_t *req, const char *buf, size_t size);
int fuse_reply_data_splice_fd(fuse_req_t *req,
const int src_fd,
const off_t pos,
const size_t size);
int fuse_reply_data(fuse_req_t *req,
char *buf,
size_t bufsize);
+62
View File
@@ -29,7 +29,69 @@ fuse_msgbuf_t *msgbuf_alloc();
fuse_msgbuf_t *msgbuf_alloc_page_aligned();
void msgbuf_free(fuse_msgbuf_t *msgbuf);
int msgbuf_ensure_pipe(fuse_msgbuf_t *msgbuf);
void msgbuf_drain_pipe(fuse_msgbuf_t *msgbuf);
void msgbuf_clear();
void msgbuf_gc();
u64 msgbuf_alloc_count();
/* Shared splice-transport plumbing used by both fuse_msgbuf.cpp and
fuse_lowlevel.cpp/fuse.cpp - factored here so the two translation
units can't independently drift on pipe sizing/logging. */
/* Below this size the extra splice/vmsplice syscalls cost more than
the copy they avoid (benchmark rationale: see src/config.cpp's
comment on the `splice` option). Shared by the read-reply gate
(fuse.cpp) and the write/splice-move gate (fuse_lowlevel.cpp) so
the two thresholds can't silently drift apart. */
constexpr u32 FUSE_SPLICE_MIN_SIZE = 131072u;
/* Read /proc/sys/fs/pipe-max-size once per process (cached) and
return it, clamped to UINT32_MAX; returns 1048576 (the historical
assumed default) if the sysctl can't be read or parsed. */
u32 fuse_effective_pipe_max();
enum class splice_fallback_tag_t
{
PIPE_MAX_READ_FALLBACK,
MSGBUF_FSETPIPE_EP,
MSGBUF_FSETPIPE_ANY,
MSGBUF_PIPE_CREATE,
REPLY_PIPE_FSETPIPE_EP,
REPLY_PIPE_FSETPIPE_ANY,
REPLY_PIPE_ENSURE,
REPLY_PIPE_CAP,
REPLY_VMSPLICE_HDR,
REPLY_FD_SPLICE,
RECEIVE_MSGBUF_PIPE,
RECEIVE_SPLICE_ERR,
REPLY_SPLICE_MOVE_WRITEV,
};
/* Warn once per process per tag so a busy mount doesn't spam
per-request logs when splice degrades to the copy path. */
void fuse_splice_fallback_log(splice_fallback_tag_t tag);
/* Size a pipe's kernel buffer via F_SETPIPE_SZ, clamping the request
to INT_MAX and retrying once at the 1MiB default on failure -
unconditionally when retry_unconditionally is set, otherwise only
when the first attempt failed with EPERM. Logs tag_any once if the
sized request failed at all, tag_ep once if every attempt failed.
Returns the granted size (>0) or -errno of the last attempt (<=0). */
int fuse_pipe_set_size(int fd,
u32 cap,
bool retry_unconditionally,
splice_fallback_tag_t tag_any,
splice_fallback_tag_t tag_ep);
/* Bounded, non-blocking pipe drain: consume exactly the bytes FIONREAD
reports as queued, reading into scratch (capped at scratch_size),
retrying on EINTR. Never blocks - the write end is assumed still
held by the caller, so a blocking read on an already-drained empty
pipe would wedge the thread forever. No-op if fd is -1. Shared by
msgbuf_drain_pipe (this file) and _ReplyPipe::drain
(fuse_lowlevel.cpp) so the receive-side and reply-side pipes can't
independently drift on drain semantics. */
void fuse_drain_pipe_fd(int fd, void *scratch, size_t scratch_size);
+6 -1
View File
@@ -2,9 +2,14 @@
#include <stdint.h>
typedef struct fuse_msgbuf_t fuse_msgbuf_t;
/* No separate "typedef struct fuse_msgbuf_t fuse_msgbuf_t;" - that's a
C idiom (a tag name isn't usable as a type name on its own in C);
in C++ the struct's tag is already a type name. */
struct fuse_msgbuf_t
{
uint32_t size;
char *mem;
int pipefd[2] = {-1,-1};
uint32_t pipe_used = 0;
uint32_t pipe_cap = 0;
};
+2
View File
@@ -2,6 +2,7 @@
#include "fuse_req_ctx.h"
#include "fuse_conn_info.hpp"
#include "fuse_msgbuf_t.h"
struct fuse_session;
@@ -12,6 +13,7 @@ struct fuse_req_t
struct fuse_session *se;
int fd;
fuse_conn_info_t conn;
fuse_msgbuf_t *msgbuf = nullptr;
unsigned int ioctl_64bit : 1;
};
+7
View File
@@ -31,6 +31,13 @@ namespace SysLog
open()
{
const char *ident = "mergerfs";
/* Deliberately NOT LOG_PERROR: that would echo every SysLog call
in the whole process (not just the splice fallback warnings) to
stderr, doubling log volume under systemd/journald setups where
stderr is also journal-captured. The splice fallback loggers
(fuse_splice_fallback_log / msgbuf_splice_fallback_log) print to
stderr themselves instead, scoping the "visible in foreground
runs" behavior to the messages that actually need it. */
const int option = (LOG_CONS|LOG_PID);
const int facility = LOG_USER;
+386
View File
@@ -0,0 +1,386 @@
/*
FUSE: Filesystem in Userspace
Copyright (C) 2010 Miklos Szeredi <miklos@szeredi.hu>
This program can be distributed under the terms of the GNU LGPLv2.
See the file COPYING.LIB
*/
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include "fuse_common.h"
#include <algorithm>
#include <array>
#include <cassert>
#include <cerrno>
#include <cstdint>
#include <cstring>
#include <fcntl.h>
#include <unistd.h>
size_t
fuse_buf_size(const struct fuse_bufvec *bufv_)
{
size_t i;
size_t size = 0;
for(i = 0; i < bufv_->count; i++)
{
if(bufv_->buf[i].size == SIZE_MAX)
size = SIZE_MAX;
else
size += bufv_->buf[i].size;
}
return size;
}
static
ssize_t
fuse_buf_write(const struct fuse_buf *dst,
size_t dst_off_,
const struct fuse_buf *src,
size_t src_off_,
size_t len_)
{
ssize_t res = 0;
size_t copied = 0;
while(len_)
{
if(dst->flags & FUSE_BUF_FD_SEEK)
res = pwrite(dst->fd,(char*)src->mem + src_off_,len_,dst->pos + dst_off_);
else
res = write(dst->fd,(char*)src->mem + src_off_,len_);
if(res == -1)
{
if(!copied)
return -errno;
break;
}
if(res == 0)
break;
copied += res;
if(!(dst->flags & FUSE_BUF_FD_RETRY))
break;
src_off_ += res;
dst_off_ += res;
len_ -= res;
}
return copied;
}
static
ssize_t
fuse_buf_read(const struct fuse_buf *dst,
size_t dst_off_,
const struct fuse_buf *src,
size_t src_off_,
size_t len_)
{
ssize_t res = 0;
size_t copied = 0;
while(len_)
{
if(src->flags & FUSE_BUF_FD_SEEK)
res = pread(src->fd,(char*)dst->mem + dst_off_,len_,src->pos + src_off_);
else
res = read(src->fd,(char*)dst->mem + dst_off_,len_);
if(res == -1)
{
if(!copied)
return -errno;
break;
}
if(res == 0)
break;
copied += res;
if(!(src->flags & FUSE_BUF_FD_RETRY))
break;
dst_off_ += res;
src_off_ += res;
len_ -= res;
}
return copied;
}
static
/* NOTE on partial-failure byte loss: when src is a one-shot,
unrereadable pipe (e.g. mergerfs's write-splice fallback), a
fuse_buf_write() failure on any chunk after the first (copied > 0
below) silently drops that chunk's already-consumed
fuse_buf_read() bytes - they were removed from the pipe but never
land at dst, and the returned "copied" count has no way to signal
the loss. This mirrors upstream libfuse's bounce-copy semantics and
is not changed here. The dominant real-world trigger in mergerfs
(an O_APPEND destination fd, which deterministically sends every
splice(2) call through this function for the whole payload) is
avoided upstream of this call: src/fuse_write_buf.cpp's
FUSE::write_buf detects O_APPEND before ever attempting splice and
routes straight to its own lossless heap-buffer fallback instead. */
ssize_t
fuse_buf_fd_to_fd(const struct fuse_buf *dst,
size_t dst_off_,
const struct fuse_buf *src,
size_t src_off_,
size_t len_)
{
std::array<char,4096> buf;
struct fuse_buf tmp =
{
.size = buf.size(),
.flags = (enum fuse_buf_flags)0,
.mem = buf.data(),
.fd = -1,
.pos = 0,
};
ssize_t res;
size_t copied = 0;
while(len_)
{
size_t this_len = std::min(tmp.size,len_);
size_t read_len;
res = fuse_buf_read(&tmp,0,src,src_off_,this_len);
if(res < 0)
{
if(!copied)
return res;
break;
}
if(res == 0)
break;
read_len = res;
res = fuse_buf_write(dst,dst_off_,&tmp,0,read_len);
if(res < 0)
{
if(!copied)
return res;
break;
}
if(res == 0)
break;
copied += res;
if((size_t)res < this_len)
break;
dst_off_ += res;
src_off_ += res;
len_ -= res;
}
return copied;
}
#if defined(__linux__)
static
ssize_t
fuse_buf_splice(const struct fuse_buf *dst,
size_t dst_off_,
const struct fuse_buf *src,
size_t src_off_,
size_t len_,
enum fuse_buf_copy_flags flags_)
{
int splice_flags = 0;
off_t *srcpos = nullptr;
off_t *dstpos = nullptr;
off_t srcpos_val;
off_t dstpos_val;
ssize_t res;
size_t copied = 0;
if(flags_ & FUSE_BUF_SPLICE_MOVE)
splice_flags |= SPLICE_F_MOVE;
if(flags_ & FUSE_BUF_SPLICE_NONBLOCK)
splice_flags |= SPLICE_F_NONBLOCK;
if(src->flags & FUSE_BUF_FD_SEEK)
{
srcpos_val = src->pos + src_off_;
srcpos = &srcpos_val;
}
if(dst->flags & FUSE_BUF_FD_SEEK)
{
dstpos_val = dst->pos + dst_off_;
dstpos = &dstpos_val;
}
while(len_)
{
res = splice(src->fd,srcpos,dst->fd,dstpos,len_,splice_flags);
if(res == -1)
{
if(copied)
break;
if(errno != EINVAL || (flags_ & FUSE_BUF_FORCE_SPLICE))
return -errno;
/* Maybe splice is not supported for this combination */
return fuse_buf_fd_to_fd(dst,dst_off_,src,src_off_,len_);
}
if(res == 0)
break;
copied += res;
if(!(src->flags & FUSE_BUF_FD_RETRY) &&
!(dst->flags & FUSE_BUF_FD_RETRY))
{
break;
}
/* The kernel advances off_in and off_out in place on every
splice() call; do NOT advance srcpos_val/dstpos_val here
(upstream semantics - a manual advance double-steps and
silently skips source data across loop iterations). */
len_ -= res;
}
return copied;
}
#endif /* defined(__linux__) */
static
ssize_t
fuse_buf_copy_one(const struct fuse_buf *dst,
size_t dst_off_,
const struct fuse_buf *src,
size_t src_off_,
size_t len_,
enum fuse_buf_copy_flags flags_)
{
bool src_is_fd = (src->flags & FUSE_BUF_IS_FD) != 0;
bool dst_is_fd = (dst->flags & FUSE_BUF_IS_FD) != 0;
if(!src_is_fd && !dst_is_fd)
{
char *dstmem = (char*)dst->mem + dst_off_;
char *srcmem = (char*)src->mem + src_off_;
if(dstmem != srcmem)
{
if(dstmem + len_ <= srcmem || srcmem + len_ <= dstmem)
memcpy(dstmem,srcmem,len_);
else
memmove(dstmem,srcmem,len_);
}
return len_;
}
else if(!src_is_fd)
{
return fuse_buf_write(dst,dst_off_,src,src_off_,len_);
}
else if(!dst_is_fd)
{
return fuse_buf_read(dst,dst_off_,src,src_off_,len_);
}
else if(flags_ & FUSE_BUF_NO_SPLICE)
{
return fuse_buf_fd_to_fd(dst,dst_off_,src,src_off_,len_);
}
else
{
#if defined(__linux__)
return fuse_buf_splice(dst,dst_off_,src,src_off_,len_,flags_);
#else
/* splice(2)/SPLICE_F_MOVE/SPLICE_F_NONBLOCK are Linux-only;
other platforms always take the portable bounce-copy path
(same behavior fuse_buf_splice itself falls back to on
Linux when splice(2) returns EINVAL). */
return fuse_buf_fd_to_fd(dst,dst_off_,src,src_off_,len_);
#endif
}
}
static
const struct fuse_buf*
fuse_bufvec_current(struct fuse_bufvec *bufv_)
{
if(bufv_->idx < bufv_->count)
return &bufv_->buf[bufv_->idx];
else
return nullptr;
}
static
bool
fuse_bufvec_advance(struct fuse_bufvec *bufv_,
size_t len_)
{
const struct fuse_buf *buf = fuse_bufvec_current(bufv_);
bufv_->off += len_;
assert(bufv_->off <= buf->size);
if(bufv_->off == buf->size)
{
assert(bufv_->idx < bufv_->count);
bufv_->idx++;
if(bufv_->idx == bufv_->count)
return false;
bufv_->off = 0;
}
return true;
}
ssize_t
fuse_buf_copy(struct fuse_bufvec *dstv_,
struct fuse_bufvec *srcv_,
enum fuse_buf_copy_flags flags_)
{
size_t copied = 0;
if(dstv_ == srcv_)
return fuse_buf_size(dstv_);
for(;;)
{
const struct fuse_buf *src = fuse_bufvec_current(srcv_);
const struct fuse_buf *dst = fuse_bufvec_current(dstv_);
size_t src_len;
size_t dst_len;
size_t len;
ssize_t res;
if(src == nullptr || dst == nullptr)
break;
src_len = src->size - srcv_->off;
dst_len = dst->size - dstv_->off;
len = std::min(src_len,dst_len);
res = fuse_buf_copy_one(dst,dstv_->off,src,srcv_->off,len,flags_);
if(res < 0)
{
if(!copied)
return res;
break;
}
copied += res;
if(!fuse_bufvec_advance(srcv_,res) ||
!fuse_bufvec_advance(dstv_,res))
break;
if((size_t)res < len)
break;
}
return copied;
}
+118 -29
View File
@@ -2326,15 +2326,78 @@ fuse_lib_open(fuse_req_t *req_,
free_path(hdr_->nodeid,fusepath);
}
/* Shared by fuse_lib_read_fd_splice's fallback path and fuse_lib_read's
own body: allocate a page-aligned msgbuf, run the plain ops.read
into it, and reply. */
static
void
fuse_lib_read_mem(fuse_req_t *req_,
const fuse_file_info_t *ffi_,
const size_t size_,
const off_t offset_)
{
fuse_msgbuf_t *msgbuf = msgbuf_alloc_page_aligned();
if(msgbuf == nullptr)
{
fuse_reply_err(req_,ENOMEM);
return;
}
int res = f.ops.read(&req_->ctx,
ffi_,
msgbuf->mem,
size_,
offset_);
if(res >= 0)
fuse_reply_data(req_,msgbuf->mem,res);
else
fuse_reply_err(req_,res);
msgbuf_free(msgbuf);
}
static
void
fuse_lib_read_fd_splice(fuse_req_t *req_,
struct fuse_in_header *hdr_,
fuse_file_info_t *ffi_,
struct fuse_read_in *arg_)
{
struct fuse_bufvec buf;
int res = f.ops.read_buf(&req_->ctx,
ffi_,
&buf,
arg_->size,
arg_->offset);
/* read_buf returning <0 means "not applicable" (e.g. O_DIRECT fd,
memory pressure): fall straight to the mem-buffer read path. It
never returns a user-visible error value. */
if(res == 0 &&
(buf.count == 1) &&
(buf.buf[0].flags & FUSE_BUF_IS_FD) &&
(buf.buf[0].flags & FUSE_BUF_FD_SEEK))
{
const int src_fd = buf.buf[0].fd;
const off_t pos = buf.buf[0].pos;
const size_t sz = buf.buf[0].size;
int sv = fuse_reply_data_splice_fd(req_,src_fd,pos,sz);
if(sv == 0 || sv == -EIO)
return;
/* sv == -1: nothing committed to the wire. Fall through to the
mem-buffer read path as if read_buf had not been set. */
}
fuse_lib_read_mem(req_,ffi_,arg_->size,arg_->offset);
}
static
void
fuse_lib_read(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
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;
@@ -2344,25 +2407,24 @@ fuse_lib_read(fuse_req_t *req_,
ffi.lock_owner = arg->lock_owner;
}
msgbuf = msgbuf_alloc_page_aligned();
if(msgbuf == NULL)
/* fd-referenced read_buf: kernel splices payload straight from the
branch fd via a pipe into /dev/fuse - data never crosses
userspace. Wire only when the mount negotiated splice_read and
the fs registered read_buf and the request size is large enough
for kernel-side page moves to beat the memcpy. Threshold = 128K:
kernel sends bs=128K reads at exactly 131072 bytes (verified in
strace), so small_read still hits the fd path; <128K skips the
two extra syscalls where they cost more than they save. */
if(f.ops.read_buf != nullptr &&
fuse_cfg.splice_read &&
arg->size >= FUSE_SPLICE_MIN_SIZE &&
(req_->conn.want & FUSE_CAP_SPLICE_READ))
{
fuse_reply_err(req_,ENOMEM);
fuse_lib_read_fd_splice(req_,hdr_,&ffi,arg);
return;
}
res = f.ops.read(&req_->ctx,
&ffi,
msgbuf->mem,
arg->size,
arg->offset);
if(res >= 0)
fuse_reply_data(req_,msgbuf->mem,res);
else
fuse_reply_err(req_,res);
msgbuf_free(msgbuf);
fuse_lib_read_mem(req_,&ffi,arg->size,arg->offset);
}
static
@@ -2371,29 +2433,56 @@ fuse_lib_write(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
int res;
char *data;
fuse_file_info_t ffi = {};
struct fuse_write_in *arg;
arg = (fuse_write_in*)fuse_hdr_arg(hdr_);
ffi.fh = arg->fh;
ffi.writepage = !!(arg->write_flags & 1);
if(req_->conn.proto_minor < 9)
{
data = ((char*)arg) + FUSE_COMPAT_WRITE_IN_SIZE;
}
else
if(req_->conn.proto_minor >= 9)
{
ffi.flags = arg->flags;
ffi.lock_owner = arg->lock_owner;
data = (char*)PARAM(arg);
}
res = f.ops.write(&req_->ctx,
&ffi,
data,
arg->size,
arg->offset);
/* Splice-capable receive left the write payload in the pipe and
staged write_in into req_->msgbuf->mem. */
if(f.ops.write_buf != nullptr &&
req_->msgbuf != nullptr &&
req_->msgbuf->pipe_used > 0)
{
struct fuse_bufvec bufv;
bufv = fuse_bufvec_fd(req_->msgbuf->pipefd[0],arg->offset,arg->size,
FUSE_BUF_IS_FD);
res = f.ops.write_buf(&req_->ctx,&ffi,&bufv,arg->offset);
/* Only clear pipe_used when the handler consumed the whole
payload: a short write leaves the remainder queued in the
pipe, and releasing a dirty pipe into the process-global
cache (fuse_msgbuf.cpp's g_cached_pipes) would corrupt the
next request staged into it. On partial
success pipe_used stays set and msgbuf_free's bounded drain
consumes the remainder. */
if(res >= 0 && (size_t)res == arg->size)
req_->msgbuf->pipe_used = 0;
}
else
{
char *data;
data = ((req_->conn.proto_minor < 9) ?
(((char*)arg) + FUSE_COMPAT_WRITE_IN_SIZE) :
(char*)PARAM(arg));
res = f.ops.write(&req_->ctx,
&ffi,
data,
arg->size,
arg->offset);
}
if(res >= 0)
fuse_reply_write(req_,res);
else
+700 -5
View File
@@ -32,6 +32,7 @@
#include <errno.h>
#include <assert.h>
#include <sys/file.h>
#include <sys/ioctl.h>
#define PARAM(inarg) (((char*)(inarg)) + sizeof(*(inarg)))
#define OFFSET_MAX 0x7fffffffffffffffLL
@@ -119,6 +120,20 @@ fuse_send_msg(const int fd_,
return 0;
}
/* ---------------------------------------------------------------------------
* splice fallback logging
*
* Every path that silently degrades splice behavior (pipe sizing
* fallbacks, sysctl read failures, splice(2) errors) logs once per
* process so operators can see why splice went quiet, without spamming
* per-request logs on a busy mount. The logger itself (and the pipe
* sizing helpers below) live in fuse_msgbuf.cpp/.hpp, shared with the
* receive-side pipe plumbing there.
* ------------------------------------------------------------------------- */
#include "syslog.hpp"
#include <array>
#define MAX_ERRNO 4095
int
@@ -420,6 +435,457 @@ fuse_reply_buf(fuse_req_t *req,
return send_reply_ok(req, buf, size);
}
/*
* Kernel 5.x removed vmsplice() of user pages into /dev/fuse directly
* (EBADF): vmsplice targets a pipe, always has. Use the classic
* pipe-bounce: vmsplice header+data pages into a per-thread pipe, then
* splice(2) that pipe into the fuse device so the kernel sees a single
* contiguous reply message.
*
* Thread-local pipe so read-reply pages are shared with the splice
* receive worker that created them; lazily sized to pipe-max.
*/
/* Thread-local reply bounce pipe. Wrapped in a tiny RAII struct
(rather than a bare thread_local int[2] + a free "close" function)
so its destructor runs automatically at thread exit and closes both
fds: a POD array has no such hook, and the previous free-function
equivalent of this destructor had zero call sites anywhere in the
tree, silently leaking 2 fds per worker thread for the life of the
process on every mount that ever serviced a >=128KiB read reply or
a splice-move reply.
ensure/ready/drain/vmspliceIn/spliceOut are methods on the struct
itself, rather than free functions taking no arguments and reaching
into the g_reply_pipe global implicitly - same shape as Mutex/
LockGuard (mutex.hpp) and FileInfo (src/fileinfo.hpp) elsewhere in
this codebase. */
struct _ReplyPipe
{
int fd[2] = {-1,-1};
u32 cap = 0;
~_ReplyPipe()
{
if(fd[0] != -1) ::close(fd[0]);
if(fd[1] != -1) ::close(fd[1]);
}
int ensure();
bool ready(size_t needed, size_t slots);
void drain();
ssize_t vmspliceIn(struct iovec *iov, int iovcnt);
ssize_t spliceOut(int dst_fd, size_t total);
};
static thread_local _ReplyPipe g_reply_pipe;
/* Everything below through _fuse_reply_data_splice_move uses
vmsplice(2)/splice(2)/SPLICE_F_MOVE/SPLICE_F_NONBLOCK/F_SETPIPE_SZ,
all Linux-only. On other platforms only the public
fuse_reply_data_splice_fd entry point needs to exist (fuse.cpp
calls it unconditionally); it always reports "nothing committed,
fall back" so callers transparently use the portable read/write
path, matching the documented behavior ("splice remains
unavailable on platforms that lack it; the copy path is the
fallback everywhere"). */
#if defined(__linux__)
int
_ReplyPipe::ensure()
{
if(fd[0] != -1)
return 0;
int rv = ::pipe2(fd,O_CLOEXEC);
if(rv == -1)
{
fd[0] = -1;
fd[1] = -1;
return -errno;
}
/* size to whatever the kernel will let an unprivileged user have.
pipe-max-size read and F_SETPIPE_SZ-with-retry are shared with
msgbuf_ensure_pipe (fuse_msgbuf.cpp) so the two can't disagree
about the pipe's real capacity. */
{
u32 pipe_max = fuse_effective_pipe_max();
/* F_SETPIPE_SZ failure is non-fatal here (unlike msgbuf_ensure_pipe):
pipe just stays default-size and splice will short-write, which
ready()'s cap check below already handles. */
rv = fuse_pipe_set_size(fd[0],pipe_max,/*retry_unconditionally_=*/true,
splice_fallback_tag_t::REPLY_PIPE_FSETPIPE_ANY,
splice_fallback_tag_t::REPLY_PIPE_FSETPIPE_EP);
cap = (rv > 0) ? (u32)rv : 65536;
}
return 0;
}
void
_ReplyPipe::drain()
{
/* Discard whatever is still queued in the bounce pipe after an
error mid-reply: without this, the NEXT fuse_reply_data on this
thread would prepend stale bytes to its fresh one, corrupting the
wire stream. */
std::array<char,65536> buf;
fuse_drain_pipe_fd(fd[0],buf.data(),buf.size());
}
/* vmsplice iov (header + payload) into the pipe. Returns -errno
* only when nothing was moved. */
ssize_t
_ReplyPipe::vmspliceIn(struct iovec *iov_,
int iovcnt_)
{
size_t done = 0;
size_t want = iov_length(iov_,iovcnt_);
while(done < want)
{
ssize_t res = vmsplice(fd[1],iov_,iovcnt_,SPLICE_F_MOVE|SPLICE_F_NONBLOCK);
if(res == -1)
{
int e = errno;
if(e == EINTR)
continue;
if(done == 0)
return -e;
return -EIO;
}
if(res == 0)
return -EIO;
done += res;
/* advance iov */
{
size_t left = res;
int i = 0;
for(; i < iovcnt_ && left > 0; )
{
if(iov_[i].iov_len <= left)
{
left -= iov_[i].iov_len;
i++;
}
else
{
iov_[i].iov_base = (char*)iov_[i].iov_base + left;
iov_[i].iov_len -= left;
left = 0;
}
}
if(i > 0)
{
memmove(&iov_[0],&iov_[i],(iovcnt_ - i) * sizeof(struct iovec));
iovcnt_ -= i;
}
}
}
return (ssize_t)done;
}
/* Drain the pipe into the fuse device. Returns -errno only when nothing
* was moved to the device; once any bytes land, they form the start of
* an atomic reply message and the caller must NOT retry with writev. */
ssize_t
_ReplyPipe::spliceOut(const int dst_fd_,
size_t total_)
{
size_t done = 0;
while(done < total_)
{
ssize_t res = ::splice(fd[0],nullptr,
dst_fd_,nullptr,
total_ - done,
SPLICE_F_MOVE);
if(res == -1)
{
int e = errno;
if(e == EINTR)
continue;
if(done == 0)
return -e;
return -EIO;
}
if(res == 0)
{
/* pipe empty but caller expected more - shouldn't happen since
we vmspliced exactly _total_. */
return -EIO;
}
done += res;
}
return (ssize_t)done;
}
/* Splice payload from a source fd (branch file) into the reply pipe,
* header first via vmsplice. Wire is untouched until the final
* pipe->/dev/fuse splice, so any failure before that is
* fallback-safe.
*
* Public splice-from-fd reply. Return semantics:
* 0 success, req consumed (freed), caller does NOT fuse_req_free
* -1 nothing committed to the wire; caller must fall back to another
* reply mechanism (or pread) and free req there
* -EIO wire committed a partial message; caller just frees req
*/
static
int
_fuse_reply_data_splice_fd(fuse_req_t *req_,
const int src_fd_,
const off_t pos_,
size_t size_);
int
fuse_reply_data_splice_fd(fuse_req_t *req_,
const int src_fd_,
const off_t pos_,
const size_t size_)
{
int rv = _fuse_reply_data_splice_fd(req_,src_fd_,pos_,size_);
if(rv == -1)
return -1;
fuse_req_free(req_);
return rv;
}
/* Number of pipe ring slots a single source segment consumes. A pipe's
capacity is enforced in whole page-sized slots, not bytes: each page
touched by a segment - including a partial first and last page -
becomes its own pipe buffer. */
static
size_t
_pipe_slots(const uintptr_t base_,
const size_t len_)
{
size_t ps = (pagesize > 0) ? pagesize : 4096;
if(len_ == 0)
return 0;
return (((base_ % ps) + len_ + ps - 1) / ps);
}
/* Shared by both reply-splice variants below: ensure the thread-local
reply pipe exists and can hold needed_ bytes in slots_ ring slots,
logging the same two fallback tags either variant would have logged
inline. */
bool
_ReplyPipe::ready(size_t needed_,
size_t slots_)
{
if(ensure() != 0)
{
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_PIPE_ENSURE);
return false;
}
/* Pipe capacity guard: we need header + payload to fit, else splice
would block with no reader on /dev/fuse advanced. cap is sized to
max pipe-max >= msgbuf but we verify explicitly. The byte check
alone is not sufficient: the kernel limits a pipe to cap/pagesize
slots and every partial page (the 16 byte header especially)
burns a whole one, so a byte-wise fitting message can still fill
the ring and block forever. */
if(needed_ > cap)
{
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_PIPE_CAP);
return false;
}
if(slots_ > (cap / ((pagesize > 0) ? pagesize : 4096)))
{
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_PIPE_CAP);
return false;
}
return true;
}
/* Shared by both reply-splice variants below: a partial write to
/dev/fuse after any bytes reached the wire poisons the connection
(unlike every other, milder fallback condition in these functions),
so - unlike those - it's always logged, not just once per process. */
static
void
_reply_splice_desync_log(const char *variant_)
{
SysLog::error("splice reply: partial write to /dev/fuse"
" ({}); connection may be desynced",variant_);
fprintf(stderr,"mergerfs: splice reply: partial write to /dev/fuse"
" (%s); connection may be desynced\n",variant_);
}
static
int
_fuse_reply_data_splice_fd(fuse_req_t *req_,
const int src_fd_,
const off_t pos_,
size_t size_)
{
struct fuse_out_header out;
struct iovec iov[1];
out.unique = req_->ctx.unique;
out.error = 0;
out.len = (u32)(sizeof(out) + size_);
if(fuse_cfg.debug)
fuse_debug_data_out(req_->ctx.unique,size_);
size_t total = sizeof(out) + size_;
size_t slots = (_pipe_slots((uintptr_t)&out,sizeof(out)) +
_pipe_slots((uintptr_t)pos_,size_));
if(!g_reply_pipe.ready(total,slots))
return -1;
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(out);
ssize_t rh = g_reply_pipe.vmspliceIn(iov,1);
if(rh != (ssize_t)sizeof(out))
{
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_VMSPLICE_HDR);
g_reply_pipe.drain();
return -1;
}
/* fd -> pipe: kernel-internal copy from page cache into pipe ring.
Short return = EOF or a transient failure; either way the wire is
still clean and the caller can fall back to a plain pread. */
{
size_t done = 0;
while(done < size_)
{
off_t off = pos_ + (off_t)done;
/* SPLICE_F_NONBLOCK applies to the pipe end only (the file read
still blocks as needed): if the slot accounting in ready()
ever under-estimates, this returns EAGAIN and we degrade to
the copy path instead of blocking forever with no reader. */
ssize_t res = ::splice(src_fd_,&off,
g_reply_pipe.fd[1],nullptr,
size_ - done,
SPLICE_F_MOVE|SPLICE_F_NONBLOCK);
if(res == -1)
{
if(errno == EINTR)
continue;
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_FD_SPLICE);
g_reply_pipe.drain();
return -1;
}
if(res == 0)
{
/* short read at EOF: drain partial payload and let the
caller reprobe with pread so ERR/EOF padding matches
the normal path. */
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_FD_SPLICE);
g_reply_pipe.drain();
return -1;
}
done += res;
}
}
ssize_t sent = g_reply_pipe.spliceOut(req_->fd,total);
if(sent == -EIO)
{
/* partial wire send: the request is poisoned, but any bytes
still queued in the bounce pipe must be drained or the next
reply on this thread would be prefixed with stale bytes. */
_reply_splice_desync_log("fd-splice read reply");
g_reply_pipe.drain();
return -EIO;
}
if(sent < 0)
{
/* nothing reached the wire; the wire is clean and the caller
may fall back to pread. Contract is {0,-1,-EIO} - never
propagate a raw -errno (the public wrapper frees the req for
any rv != -1, and callers only guard 0/-EIO/-1). */
g_reply_pipe.drain();
return -1;
}
return 0;
}
static
int
_fuse_reply_data_splice_move(fuse_req_t *req,
struct iovec *iov_,
int iovcnt_)
{
size_t want = iov_length(iov_,iovcnt_);
size_t slots = 0;
for(int i = 0; i < iovcnt_; i++)
slots += _pipe_slots((uintptr_t)iov_[i].iov_base,iov_[i].iov_len);
if(!g_reply_pipe.ready(want,slots))
return -1;
ssize_t total = g_reply_pipe.vmspliceIn(iov_,iovcnt_);
if(total < 0)
{
g_reply_pipe.drain();
return -1;
}
if((size_t)total != want)
{
g_reply_pipe.drain();
return -EIO;
}
ssize_t n = g_reply_pipe.spliceOut(req->fd,(size_t)total);
if(n < 0)
{
g_reply_pipe.drain();
/* -EIO when a partial message reached the wire (poisoned req);
anything else means nothing moved - caller may fall back to
writev. Never a raw -errno: the fallthrough caller only
distinguishes 0 / -EIO / fallback. */
if(n == -EIO)
_reply_splice_desync_log("splice-move reply");
return (n == -EIO) ? -EIO : -1;
}
if((size_t)n != (size_t)total)
{
g_reply_pipe.drain();
return -EIO;
}
/* All sent; reply complete from kernel's point of view. */
return 0;
}
#else /* !defined(__linux__) */
int
fuse_reply_data_splice_fd(fuse_req_t *req_,
const int src_fd_,
const off_t pos_,
const size_t size_)
{
(void)req_; (void)src_fd_; (void)pos_; (void)size_;
/* -1: nothing committed to the wire; caller falls back to a plain
pread + fuse_reply_data. */
return -1;
}
#endif /* defined(__linux__) */
int
fuse_reply_data(fuse_req_t *req,
char *buf_,
@@ -435,11 +901,49 @@ fuse_reply_data(fuse_req_t *req,
iov[1].iov_len = bufsize_;
out.unique = req->ctx.unique;
out.error = 0;
out.error = 0;
/* fuse_send_msg computes out->len from iov_len on entry; when we take
the pipe-bounce path we bypass it, so set len here too. */
out.len = (u32)(sizeof(out) + bufsize_);
if(fuse_cfg.debug)
fuse_debug_data_out(req->ctx.unique,bufsize_);
/* Pipe-bounce splice_move reply: page-aligned payload moved
zero-copy into the kernel's reply queue. Gated at 128 KiB (the
same threshold as the fd-splice reply path); sub-threshold
replies stay on writev - measured: 4K payloads lose ~30% on the
pipe path, 128K+ gains ~14-19%. Returns:
0 - success (reply sent)
-EIO - partial move/transmit: the request is poisoned, do NOT
fall back to writev or the kernel will see a duplicated prefix
< other -errno - nothing moved; caller may fall back to writev
*/
#if defined(__linux__)
if(bufsize_ >= FUSE_SPLICE_MIN_SIZE && fuse_cfg.splice_move)
{
/* _ReplyPipe::vmspliceIn destructively advances its iov on a
partial vmsplice; the writev fallthrough below must see the
ORIGINAL header+payload, so pass a copy, never our stack
array. */
struct iovec spiov[2] = {iov[0], iov[1]};
res = _fuse_reply_data_splice_move(req,spiov,2);
if(res == 0)
{
fuse_req_free(req);
return 0;
}
if(res == -EIO)
{
fuse_req_free(req);
return res;
}
fuse_splice_fallback_log(splice_fallback_tag_t::REPLY_SPLICE_MOVE_WRITEV);
/* fall through to writev */
}
#endif /* defined(__linux__) */
res = fuse_send_msg(req->fd,iov,2);
fuse_req_free(req);
@@ -1000,6 +1504,12 @@ do_init(fuse_req_t *req,
f.conn.capable |= FUSE_CAP_BIG_WRITES;
if(inargflags & FUSE_DONT_MASK)
f.conn.capable |= FUSE_CAP_DONT_MASK;
if(inargflags & FUSE_SPLICE_WRITE)
f.conn.capable |= FUSE_CAP_SPLICE_WRITE;
if(inargflags & FUSE_SPLICE_MOVE)
f.conn.capable |= FUSE_CAP_SPLICE_MOVE;
if(inargflags & FUSE_SPLICE_READ)
f.conn.capable |= FUSE_CAP_SPLICE_READ;
if(inargflags & FUSE_POSIX_ACL)
f.conn.capable |= FUSE_CAP_POSIX_ACL;
if(inargflags & FUSE_CACHE_SYMLINKS)
@@ -1074,6 +1584,12 @@ do_init(fuse_req_t *req,
outargflags |= FUSE_BIG_WRITES;
if(f.conn.want & FUSE_CAP_DONT_MASK)
outargflags |= FUSE_DONT_MASK;
if(f.conn.want & FUSE_CAP_SPLICE_WRITE)
outargflags |= FUSE_SPLICE_WRITE;
if(f.conn.want & FUSE_CAP_SPLICE_MOVE)
outargflags |= FUSE_SPLICE_MOVE;
if(f.conn.want & FUSE_CAP_SPLICE_READ)
outargflags |= FUSE_SPLICE_READ;
if(f.conn.want & FUSE_CAP_POSIX_ACL)
outargflags |= FUSE_POSIX_ACL;
if(f.conn.want & FUSE_CAP_CACHE_SYMLINKS)
@@ -1606,6 +2122,169 @@ fuse_ll_buf_receive_read(struct fuse_session *se_,
return rv;
}
/*
* read(2)-loop that reads exactly 'len_' bytes from 'fd_' into 'buf_'.
* Returns 0 on success, -errno on error.
*/
static
int
fuse_ll_read_full(const int fd_,
char *buf_,
const u32 len_)
{
u32 done = 0;
while(done < len_)
{
ssize_t rv = read(fd_,buf_ + done,len_ - done);
if(rv == -1)
{
if(errno == EINTR)
continue;
return -errno;
}
if(rv == 0)
return -EIO;
done += rv;
}
return 0;
}
/* splice(2)/SPLICE_F_MOVE are Linux-only; this whole receive path
(and its selection in fuse_lowlevel_new_common below) is compiled
only on Linux. Other platforms always use the portable
fuse_ll_buf_receive_read receive function. */
#if defined(__linux__)
static
int
fuse_ll_buf_receive_splice(struct fuse_session *se_,
int fd_,
fuse_msgbuf_t *msgbuf_)
{
if(!f.got_init)
return fuse_ll_buf_receive_read(se_,fd_,msgbuf_);
/* fuse_init may have disabled splice at INIT time (e.g. sysctl
pipe-max-size too small). The receive fn was selected before INIT
ran, so re-check here and take the plain read path cleanly
rather than failing per-msgbuf into the fallback. */
if(!fuse_cfg.splice_read && !fuse_cfg.splice_write)
return fuse_ll_buf_receive_read(se_,fd_,msgbuf_);
if(msgbuf_ensure_pipe(msgbuf_) != 0)
{
fuse_splice_fallback_log(splice_fallback_tag_t::RECEIVE_MSGBUF_PIPE);
return fuse_ll_buf_receive_read(se_,fd_,msgbuf_);
}
/* One pipe-limited read = one whole FUSE request (pipe sized >= max
request by the max_pages clamp in fuse_init). */
ssize_t splice_n = splice(fd_,nullptr,msgbuf_->pipefd[1],nullptr,msgbuf_->size,SPLICE_F_MOVE);
if(splice_n == -1)
{
int err = errno;
if(err == EINTR)
return -EINTR;
/* splice(2) not usable on this combination; fall back to the
read path for this message only. fuse_cfg.splice_read/write
are deliberately left set - a single failure here (e.g. a
transient ENOMEM) should not permanently disable splice for
the whole mount; if the failure is systemic every subsequent
message will simply retry and fall back the same way. */
fuse_splice_fallback_log(splice_fallback_tag_t::RECEIVE_SPLICE_ERR);
return fuse_ll_buf_receive_read(se_,fd_,msgbuf_);
}
if(splice_n < (int)sizeof(struct fuse_in_header))
{
fprintf(stderr,"fuse: short splice from fuse device: %zu < %zu\n",
splice_n,sizeof(struct fuse_in_header));
return -EIO;
}
/* Pull the fixed-size header back into memory. */
ssize_t rv = fuse_ll_read_full(msgbuf_->pipefd[0],
msgbuf_->mem,
sizeof(struct fuse_in_header));
if(rv != 0)
return (int)rv;
struct fuse_in_header *in = (struct fuse_in_header*)msgbuf_->mem;
/* Header sanity: in->len <= what splice actually moved. */
if(in->len > (u32)splice_n)
{
fprintf(stderr,"fuse: splice header size %u exceeds move %zu\n",
in->len,splice_n);
return -EIO;
}
/* Header sanity: in->len must be at least large enough to hold the
fixed fuse_in_header, or the subtraction below underflows (u32)
into a huge value that would then be used as a read length -
draining whatever's left in the pipe and blocking the worker
thread forever on the now-empty pipe (the write end stays open,
so read() blocks rather than returning EOF). A well-behaved
kernel never sends this; guard anyway since this path (unlike the
plain read receive) does arithmetic on in->len before the generic
opcode dispatch ever sees it. */
if(in->len < sizeof(struct fuse_in_header))
{
fprintf(stderr,"fuse: splice header size %u smaller than fuse_in_header\n",
in->len);
return -EIO;
}
u32 hdrlen = in->len - sizeof(struct fuse_in_header);
if(in->opcode == FUSE_WRITE && fuse_cfg.splice_write)
{
/* Also copy the write_in header out of the pipe so fuse_lib_write
can parse it without touching the pipe. */
u32 write_hdr = sizeof(struct fuse_write_in);
/* Same underflow guard as above, sized for this opcode: a
WRITE request must be at least header + fuse_write_in, or
both this read (fixed write_hdr length, could ask for more
than the pipe will ever hold) and the payload computation
below (which would underflow) are wrong. */
if(hdrlen < write_hdr)
{
fprintf(stderr,"fuse: splice WRITE request too short for"
" fuse_write_in: %u < %u\n",
hdrlen,write_hdr);
return -EIO;
}
int wirv = fuse_ll_read_full(msgbuf_->pipefd[0],
msgbuf_->mem + sizeof(struct fuse_in_header),
write_hdr);
if(wirv != 0)
return wirv;
/* Payload stays in the pipe for write_buf zero-copy. */
u32 payload = in->len - sizeof(struct fuse_in_header) - write_hdr;
msgbuf_->pipe_used = payload;
return (int)splice_n;
}
/* Non-write path: move the rest of the request out of the pipe and
into the msgbuf so downstream processing is identical to read(). */
rv = fuse_ll_read_full(msgbuf_->pipefd[0],
msgbuf_->mem + sizeof(struct fuse_in_header),
hdrlen);
if(rv != 0)
return (int)rv;
msgbuf_->pipe_used = 0;
return in->len;
}
#endif /* defined(__linux__) */
static
void
fuse_ll_buf_process_read(struct fuse_session *se_,
@@ -1637,6 +2316,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_,
req->se = se_;
req->fd = fd_;
req->ioctl_64bit = 0;
req->msgbuf = (fuse_msgbuf_t*)msgbuf_;
err = ENOSYS;
if(in->opcode >= FUSE_MAXOPS)
@@ -1683,6 +2363,7 @@ fuse_ll_buf_process_read_init(struct fuse_session *se_,
req->ctx.umask = 0;
req->se = se_;
req->fd = fd_;
req->msgbuf = (fuse_msgbuf_t*)msgbuf_;
err = EIO;
if(in->opcode != FUSE_INIT)
@@ -1726,10 +2407,24 @@ fuse_lowlevel_new_common(struct fuse_args *args,
memcpy(&f.op,op,op_size);
f.owner = getuid();
se = fuse_session_new(&f,
(void*)fuse_ll_buf_receive_read,
(void*)fuse_ll_buf_process_read_init,
(void*)fuse_ll_destroy);
{
/* Splice-capable receive needs a pipe buffer per msgbuf.
fuse_ll_buf_receive_splice only exists on Linux (see its
definition above); other platforms always take the portable
read() receive path. */
#if defined(__linux__)
if(fuse_cfg.splice_read || fuse_cfg.splice_write)
se = fuse_session_new(&f,
(void*)fuse_ll_buf_receive_splice,
(void*)fuse_ll_buf_process_read_init,
(void*)fuse_ll_destroy);
else
#endif
se = fuse_session_new(&f,
(void*)fuse_ll_buf_receive_read,
(void*)fuse_ll_buf_process_read_init,
(void*)fuse_ll_destroy);
}
if(!se)
goto out_free;
+471
View File
@@ -15,16 +15,88 @@
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <atomic>
#include "fuse_msgbuf.hpp"
#include "fatal.hpp"
#include "fuse.h"
#include "fuse_kernel.h"
#include "syslog.hpp"
/* Warn-once-per-process-per-tag logger shared with fuse_lowlevel.cpp
(declared in fuse_msgbuf.hpp) so the two translation units don't
each hand-roll their own copy of this mechanism. Each tag belongs
to one of two "domains" - the msgbuf/receive-side pipe plumbing in
this file, or the reply/splice-move plumbing in fuse_lowlevel.cpp -
which only differ in the fixed prefix/description text; the tag
name itself (embedded in both lines) still identifies the exact
call site. */
struct SpliceFallbackInfo
{
const char *name;
bool msgbuf_domain;
};
static constexpr SpliceFallbackInfo SPLICE_FALLBACK_INFO[] =
{
{ "pipe-max-read-fallback", true },
{ "msgbuf-fsetpipe-ep", true },
{ "msgbuf-fsetpipe-any", true },
{ "msgbuf-pipe-create", true },
{ "reply-pipe-fsetpipe-ep", false },
{ "reply-pipe-fsetpipe-any", false },
{ "reply-pipe-ensure", false },
{ "reply-pipe-cap", false },
{ "reply-vmsplice-hdr", false },
{ "reply-fd-splice", false },
{ "receive-msgbuf-pipe", false },
{ "receive-splice-err", false },
{ "reply-splice-move-writev", false },
};
void
fuse_splice_fallback_log(splice_fallback_tag_t tag_)
{
static std::atomic<bool> seen[sizeof(SPLICE_FALLBACK_INFO) /
sizeof(SPLICE_FALLBACK_INFO[0])];
const SpliceFallbackInfo &info = SPLICE_FALLBACK_INFO[(size_t)tag_];
bool expected = false;
if(!seen[(size_t)tag_].compare_exchange_strong(expected,true))
return;
const char *prefix = (info.msgbuf_domain ?
"msgbuf pipe fallback" : "splice fallback");
const char *description = (info.msgbuf_domain ?
"splice receive will use a smaller/default pipe" :
"splice I/O will use the copy path for this mount");
SysLog::warning("{} [{}]: {}",prefix,info.name,description);
/* Explicit stderr echo (rather than syslog's LOG_PERROR, which
would apply to every log message in the process): keeps this
one class of message visible on a foreground (-f) run without
doubling unrelated log output under systemd/journald. */
fprintf(stderr,"mergerfs: %s [%s]: %s\n",prefix,info.name,description);
}
#include "mutex.hpp"
#include "objpool.hpp"
#include <sys/ioctl.h>
#include <fcntl.h>
#include <unistd.h>
#include <algorithm>
#include <cassert>
#include <cerrno>
#include <charconv>
#include <climits>
#include <cstdio>
#include <cstdlib>
@@ -35,6 +107,11 @@ static u64 g_bufsize = 0;
fuse_in_header+fuse_write_in alignment */
static constexpr u64 MSGBUF_OVERHEAD_PAGES = 2ULL;
/* Guards g_cached_pipes (defined below, near its struct types.) Declared
here (ahead of _constructor(), which initializes it) rather than
next to g_cached_pipes itself. */
static mutex_t g_cached_pipes_mutex;
static
__attribute__((constructor))
void
@@ -51,6 +128,8 @@ _constructor()
< g_pagesize);
msgbuf_set_bufsize(FUSE_DEFAULT_MAX_MAX_PAGES);
mutex_init(g_cached_pipes_mutex);
}
struct PageAlignedAllocator
@@ -80,6 +159,16 @@ struct ShouldPoolMsgbuf
static ObjPool<fuse_msgbuf_t, PageAlignedAllocator, ShouldPoolMsgbuf> g_msgbuf_pool;
static
void
_msgbuf_pipe_init(fuse_msgbuf_t *msgbuf_)
{
msgbuf_->pipefd[0] = -1;
msgbuf_->pipefd[1] = -1;
msgbuf_->pipe_used = 0;
msgbuf_->pipe_cap = 0;
}
static
void
_msgbuf_page_align(fuse_msgbuf_t *msgbuf_)
@@ -87,6 +176,8 @@ _msgbuf_page_align(fuse_msgbuf_t *msgbuf_)
msgbuf_->mem = (char*)msgbuf_;
msgbuf_->mem += g_pagesize;
msgbuf_->size = (g_bufsize - g_pagesize);
_msgbuf_pipe_init(msgbuf_);
}
static
@@ -98,6 +189,358 @@ _msgbuf_write_align(fuse_msgbuf_t *msgbuf_)
msgbuf_->mem -= sizeof(struct fuse_in_header);
msgbuf_->mem -= sizeof(struct fuse_write_in);
msgbuf_->size = (g_bufsize - g_pagesize);
_msgbuf_pipe_init(msgbuf_);
}
/*
* The ObjPool value-initializes msgbufs on every alloc, which would
* clobber live pipe fds carried by a pooled buffer, so pipes are not
* stored across free/alloc cycles. Instead released pipes are stashed
* in a small process-global cache keyed by capacity so the next
* msgbuf_ensure_pipe() usually avoids pipe2(2) and
* fcntl(F_SETPIPE_SZ) entirely.
*
* Global rather than thread-local: with the opt-in process-thread
* pool, msgbuf_free() runs on worker threads while msgbuf_ensure_pipe()
* runs on read threads - a thread-local cache can never hand a pipe
* across that boundary, and every request would pay pipe2(2) +
* F_SETPIPE_SZ + close. The cache is guarded by g_cached_pipes_mutex
* (mutex_lockguard, same idiom as f.lock in fuse.cpp) - the ops it
* protects are O(1) and vastly cheaper than the syscalls they avoid.
*/
struct cached_pipe_t
{
int fd0{-1};
int fd1{-1};
u32 cap{0};
};
struct cached_pipes_t
{
static constexpr u32 MAX = 16;
u32 count{0};
cached_pipe_t entries[MAX];
};
static cached_pipes_t g_cached_pipes;
static
u32
_read_pipe_max_size(void)
{
int fd = ::open("/proc/sys/fs/pipe-max-size",O_RDONLY | O_CLOEXEC);
if(fd == -1)
{
fuse_splice_fallback_log(splice_fallback_tag_t::PIPE_MAX_READ_FALLBACK);
return 1048576;
}
char buf[32];
ssize_t rv = ::read(fd,buf,sizeof(buf));
::close(fd);
if(rv <= 0)
{
fuse_splice_fallback_log(splice_fallback_tag_t::PIPE_MAX_READ_FALLBACK);
return 1048576;
}
unsigned long v;
auto [end,ec] = std::from_chars(buf,buf + rv,v);
if(ec != std::errc() || v == 0)
{
fuse_splice_fallback_log(splice_fallback_tag_t::PIPE_MAX_READ_FALLBACK);
return 1048576;
}
/* Clamp rather than treat as unreadable: a legitimately huge
sysctl (e.g. an admin-set multi-GiB pipe-max-size) must not
be mistaken for "couldn't read it" and silently replaced by
the small 1MiB default below. */
return (v > UINT32_MAX) ? UINT32_MAX : (u32)v;
}
u32
fuse_effective_pipe_max()
{
static const u32 max = _read_pipe_max_size();
return max;
}
/* Shared by msgbuf_ensure_pipe (this file) and _reply_pipe_ensure
(fuse_lowlevel.cpp): try F_SETPIPE_SZ at cap_, retrying once at the
1MiB default on failure. tag_any_ is logged whenever the sized
request fails at all (even if the retry then succeeds); tag_ep_ is
logged only if every attempt fails. */
int
fuse_pipe_set_size(int fd_,
u32 cap_,
bool retry_unconditionally_,
splice_fallback_tag_t tag_any_,
splice_fallback_tag_t tag_ep_)
{
int err = 0;
/* F_SETPIPE_SZ takes a signed int; a cap of 2^31 or higher (reachable
when pipe-max-size and the negotiated fuse-msg-size both pass
~2GiB, e.g. large-page hosts) would sign-extend into a negative
fcntl() argument below. Clamp to INT_MAX rather than cast blindly. */
if(cap_ > (u32)INT_MAX)
cap_ = (u32)INT_MAX;
int rv = ::fcntl(fd_,F_SETPIPE_SZ,(int)cap_);
if(rv == -1)
{
/* Capture immediately: fuse_splice_fallback_log() below can
itself clobber errno before it's read again further down. */
err = -errno;
fuse_splice_fallback_log(tag_any_);
if(retry_unconditionally_ || err == -EPERM)
{
rv = ::fcntl(fd_,F_SETPIPE_SZ,1048576);
if(rv == -1)
err = -errno; /* re-capture: this is a different syscall */
}
if(rv <= 0)
fuse_splice_fallback_log(tag_ep_);
}
return (rv > 0) ? rv : err;
}
/* Shared failure-cleanup for msgbuf_ensure_pipe: close the fds when a
pipe was actually created (fd0_ != -1), reset the msgbuf's pipe
state, and return rv_ (the errno-or-fixed-code the caller wants to
propagate). */
static
int
_msgbuf_ensure_pipe_fail(fuse_msgbuf_t *msgbuf_,
int fd0_,
int fd1_,
int rv_)
{
if(fd0_ != -1)
{
::close(fd0_);
::close(fd1_);
}
msgbuf_->pipefd[0] = -1;
msgbuf_->pipefd[1] = -1;
msgbuf_->pipe_cap = 0;
return rv_;
}
int
msgbuf_ensure_pipe(fuse_msgbuf_t *msgbuf_)
{
#if !defined(__linux__)
/* F_SETPIPE_SZ (used below to size the pipe to fit a whole splice
receive) is Linux-only. Without it a splice receive can't be
reliably sized, so fail unconditionally here and let every
caller fall back to the portable read() receive path, exactly
as documented ("splice remains unavailable on platforms that
lack it; the copy path is the fallback everywhere"). */
(void)msgbuf_;
return -ENOTSUP;
#else
u32 cap = (u32)std::min((u64)g_bufsize,(u64)fuse_effective_pipe_max());
/* F_SETPIPE_SZ takes a signed int; a cap of 2^31 or higher (reachable
when pipe-max-size and the negotiated fuse-msg-size both pass
~2GiB, e.g. large-page hosts) would sign-extend into a negative
fcntl() argument below. Clamp to INT_MAX rather than cast blindly -
mirrors the identical clamp inside fuse_pipe_set_size, but the
cache lookup just below also needs the clamped value. */
if(cap > (u32)INT_MAX)
cap = (u32)INT_MAX;
if(msgbuf_->pipefd[0] != -1)
return 0;
/* try global cache first */
{
mutex_lockguard(g_cached_pipes_mutex);
for(u32 i = 0; i < g_cached_pipes.count; i++)
{
cached_pipe_t *cp = &g_cached_pipes.entries[i];
if(cp->cap >= cap)
{
msgbuf_->pipefd[0] = cp->fd0;
msgbuf_->pipefd[1] = cp->fd1;
msgbuf_->pipe_cap = cp->cap;
g_cached_pipes.entries[i] = g_cached_pipes.entries[g_cached_pipes.count - 1];
g_cached_pipes.count--;
/* mutex_lockguard's unlock is deferred to scope exit, which
this return still triggers (stack unwinding runs it
regardless of which nested block the return sits in). */
return 0;
}
}
}
int rv = ::pipe2(msgbuf_->pipefd,O_CLOEXEC);
if(rv == -1)
{
int err = -errno; /* capture before the log call: SysLog::warning
-> syslog()/LOG_PERROR can make its own
syscalls (connect/sendto/write) and clobber
errno first. */
fuse_splice_fallback_log(splice_fallback_tag_t::MSGBUF_PIPE_CREATE);
msgbuf_->pipefd[0] = -1;
msgbuf_->pipefd[1] = -1;
return err;
}
/* Only retry unconditionally on EPERM: unlike the reply pipe
(fuse_lowlevel.cpp), a non-EPERM failure here (e.g. EINVAL on
an already-too-large ask) should not mask itself with a 1MiB
retry that might still be wrong for this request. */
rv = fuse_pipe_set_size(msgbuf_->pipefd[0],cap,/*retry_unconditionally_=*/false,
splice_fallback_tag_t::MSGBUF_FSETPIPE_ANY,
splice_fallback_tag_t::MSGBUF_FSETPIPE_EP);
if(rv <= 0)
return _msgbuf_ensure_pipe_fail(msgbuf_,msgbuf_->pipefd[0],msgbuf_->pipefd[1],rv);
/* Record what F_SETPIPE_SZ actually granted, not what we asked
for: after an EPERM->1MiB retry the real capacity can be
smaller than the ask, and the process-global pipe cache must
not serve this pipe for a request it cannot hold (a short
receive splice would then fail the in->len check and kill the
worker thread). */
msgbuf_->pipe_cap = (u32)rv;
/* The EPERM->1MiB retry (or a cache hit sized for a smaller
g_bufsize than this msgbuf now needs) can grant a pipe smaller
than "cap" - what THIS request actually requires to hold one
whole message. Handing that pipe back as success would let the
caller splice from /dev/fuse into it, come up short
(splice_n < in->len), and return -EIO, which fuse_loop.cpp's
_retriable_receive_error does not treat as retriable: that
tears down the entire session over one thread's transient
pipe-sizing shortfall. Fail HERE instead, before any /dev/fuse
data has been touched, so the caller falls back to the plain
read() receive for this message like every other
msgbuf_ensure_pipe failure already does. */
if(msgbuf_->pipe_cap < cap)
{
fuse_splice_fallback_log(splice_fallback_tag_t::MSGBUF_FSETPIPE_EP);
return _msgbuf_ensure_pipe_fail(msgbuf_,msgbuf_->pipefd[0],msgbuf_->pipefd[1],-ENOSPC);
}
return 0;
#endif /* defined(__linux__) */
}
void
fuse_drain_pipe_fd(int fd_,
void *scratch_,
size_t scratch_size_)
{
if(fd_ == -1)
return;
for(;;)
{
int avail;
if(::ioctl(fd_,FIONREAD,&avail) == -1)
break;
if(avail <= 0)
break;
ssize_t rv = ::read(fd_,scratch_,std::min((size_t)avail,scratch_size_));
if(rv == -1)
{
if(errno == EINTR)
continue;
break;
}
if(rv == 0)
break;
}
}
void
msgbuf_drain_pipe(fuse_msgbuf_t *msgbuf_)
{
/* This is the cleanup path for failed spliced writes (ENOSPC etc.),
where the payload may have been fully or partially consumed
already. */
fuse_drain_pipe_fd(msgbuf_->pipefd[0],msgbuf_->mem,msgbuf_->size);
msgbuf_->pipe_used = 0;
}
static
void
_msgbuf_release_pipe(fuse_msgbuf_t *msgbuf_)
{
if(msgbuf_->pipefd[0] == -1)
return;
/* Cache bookkeeping happens under lock; the close() below (taken only
on cache overflow) deliberately happens AFTER the lock is released,
so the mutex is never held across a syscall. */
bool cached = false;
{
mutex_lockguard(g_cached_pipes_mutex);
if(g_cached_pipes.count < cached_pipes_t::MAX)
{
cached_pipe_t *cp = &g_cached_pipes.entries[g_cached_pipes.count++];
cp->fd0 = msgbuf_->pipefd[0];
cp->fd1 = msgbuf_->pipefd[1];
/* cache the capacity F_SETPIPE_SZ actually granted: every
success path through msgbuf_ensure_pipe (cache hit or fresh
fcntl) sets pipe_cap nonzero before pipefd[0] goes valid, so
there is no "no fcntl ever ran" case to fall back for here. */
cp->cap = msgbuf_->pipe_cap;
cached = true;
}
}
if(!cached)
{
::close(msgbuf_->pipefd[0]);
::close(msgbuf_->pipefd[1]);
}
msgbuf_->pipefd[0] = -1;
msgbuf_->pipefd[1] = -1;
}
/* Close and drop every pipe currently sitting in the process-global
cache. Used by msgbuf_clear() so the "gc" control-file command (the
admin-facing "shrink resource usage now" entry point) actually
reclaims the cache's fds/kernel pipe buffers instead of only
clearing the msgbuf pool - previously the cache was documented as
"reclaimed ... at process exit", but no such reclamation code
existed anywhere, so in practice it was only ever trimmed by the
count>=MAX overflow-close path inside _msgbuf_release_pipe. */
static
void
_cached_pipes_clear(void)
{
cached_pipes_t local;
/* Snapshot-and-clear under lock, close() (a syscall) after releasing
it - same discipline as _msgbuf_release_pipe above. */
{
mutex_lockguard(g_cached_pipes_mutex);
local = g_cached_pipes;
g_cached_pipes.count = 0;
}
for(u32 i = 0; i < local.count; i++)
{
::close(local.entries[i].fd0);
::close(local.entries[i].fd1);
}
}
typedef void (*msgbuf_setup_func_t)(fuse_msgbuf_t*);
@@ -135,6 +578,17 @@ msgbuf_free(fuse_msgbuf_t *msgbuf_)
if(msgbuf_ == nullptr)
return;
/* Drain whenever a pipe exists, not only when pipe_used says bytes
remain: receive-path error returns (short splice, header read
failure, in->len sanity) leave queued bytes with pipe_used == 0,
and releasing a dirty pipe into the global cache would corrupt
the next request staged into it. msgbuf_drain_pipe is a
FIONREAD-bounded no-op on a clean pipe. */
if(msgbuf_->pipefd[0] != -1)
msgbuf_drain_pipe(msgbuf_);
_msgbuf_release_pipe(msgbuf_);
g_msgbuf_pool.free_size(msgbuf_,g_bufsize);
}
@@ -159,6 +613,14 @@ msgbuf_set_bufsize(const u64 size_in_pages_)
if(new_bufsize != g_bufsize)
{
g_bufsize = new_bufsize;
/* Pool nodes are keyed by alloc size; a stale-size node must
never be reused for the new bufsize, so drop the whole pool.
(Pooled msgbufs never hold pipes: msgbuf_free always runs
_msgbuf_release_pipe first, moving any pipe into the
process-global cache or closing it. Cached pipes are keyed
by granted capacity and remain usable across a bufsize
change - msgbuf_ensure_pipe's first-fit only accepts
cp->cap >= cap.) */
g_msgbuf_pool.clear();
}
}
@@ -172,6 +634,14 @@ msgbuf_alloc_count()
void
msgbuf_gc()
{
/* gc() deallocates objects inside ObjPool; any pipe a msgbuf held
was already released to the process-global cache (or closed) by
msgbuf_free's _msgbuf_release_pipe before the node returned to
the pool. Nothing to reclaim here beyond delegating. The
process-global cache itself (up to 16 pipes / 32 fds) is left
alone by this lighter-weight periodic gc() - it survives across
threads by design - but IS reclaimed by _cached_pipes_clear() via
the heavier, explicitly-invoked msgbuf_clear(). */
g_msgbuf_pool.gc();
}
@@ -179,4 +649,5 @@ void
msgbuf_clear()
{
g_msgbuf_pool.clear();
_cached_pipes_clear();
}