mirror of
https://github.com/trapexit/mergerfs.git
synced 2026-10-26 19:01:56 -04:00
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>
386 lines
9 KiB
C++
386 lines
9 KiB
C++
/*
|
|
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;
|
|
}
|