rt/mergerfs
1
0
Fork 0
mirror of https://github.com/trapexit/mergerfs.git synced 2026-10-26 02:48:59 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions
mergerfs/vendored/libfuse/lib/fuse_lowlevel.cpp
Antonio S. Musumeci 54166afe08 splice
2026-10-10 14:32:40 -04:00

2250 lines
54 KiB
C++
Raw Blame History

/*
FUSE: Filesystem in Userspace
Copyright (C) 2001-2007 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 "mutex.hpp"
#include "debug.hpp"
#include "fatal.hpp"
#include "fmt/core.h"
#include "fuse_cfg.hpp"
#include "fuse_i.hpp"
#include "fuse_kernel.h"
#include "fuse_msgbuf.hpp"
#include "fuse_opt.h"
#include "fuse_pollhandle.h"
#include "stat_utils.h"
#include "syslog.hpp"
#include <atomic>
#if defined(__linux__)
#include <fcntl.h>
#include <sys/ioctl.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include <stddef.h>
#include <string.h>
#include <unistd.h>
#include <limits.h>
#include <errno.h>
#include <assert.h>
#include <sys/file.h>
#define PARAM(inarg) (((char*)(inarg)) + sizeof(*(inarg)))
#define OFFSET_MAX 0x7fffffffffffffffLL
#define container_of(ptr, type, member) ({ \
const decltype( ((type*)0)->member ) *__mptr = (ptr); \
(type *)( (char*)__mptr - offsetof(type,member) );})
static size_t pagesize;
struct fuse_ll
{
struct fuse_lowlevel_ops op;
uid_t owner;
fuse_conn_info_t conn;
mutex_t lock;
int got_init;
int got_destroy;
uint64_t notify_ctr;
struct fuse_notify_req notify_list;
};
static fuse_ll f = {};
static
__attribute__((constructor))
void
fuse_ll_constructor(void)
{
pagesize = sysconf(_SC_PAGESIZE);
if(pagesize <= 0)
fatal::abort("pagesize query failed - {}",strerror(errno));
}
static
void
convert_stat(const struct stat *stbuf_,
struct fuse_attr *attr_)
{
attr_->ino = stbuf_->st_ino;
attr_->mode = stbuf_->st_mode;
attr_->nlink = stbuf_->st_nlink;
attr_->uid = stbuf_->st_uid;
attr_->gid = stbuf_->st_gid;
attr_->rdev = stbuf_->st_rdev;
attr_->size = stbuf_->st_size;
attr_->blksize = stbuf_->st_blksize;
attr_->blocks = stbuf_->st_blocks;
attr_->atime = stbuf_->st_atime;
attr_->mtime = stbuf_->st_mtime;
attr_->ctime = stbuf_->st_ctime;
attr_->atimensec = ST_ATIM_NSEC(stbuf_);
attr_->mtimensec = ST_MTIM_NSEC(stbuf_);
attr_->ctimensec = ST_CTIM_NSEC(stbuf_);
}
static
size_t
iov_length(const struct iovec *iov,
size_t count)
{
size_t seg;
size_t ret = 0;
for(seg = 0; seg < count; seg++)
ret += iov[seg].iov_len;
return ret;
}
static
int
fuse_send_msg(const int fd_,
struct iovec *iov,
int count)
{
int rv;
struct fuse_out_header *out = (fuse_out_header*)iov[0].iov_base;
out->len = iov_length(iov, count);
rv = writev(fd_,iov,count);
if(rv == -1)
return -errno;
return 0;
}
#define MAX_ERRNO 4095
int
fuse_send_reply_iov_nofree(fuse_req_t *req,
int error,
struct iovec *iov,
int count)
{
struct fuse_out_header out;
if(error > 0)
error = -error;
if(error <= -MAX_ERRNO)
{
fprintf(stderr,"fuse: bad error value: %i\n",error);
error = -ERANGE;
}
out.unique = req->ctx.unique;
out.error = error;
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(struct fuse_out_header);
return fuse_send_msg(req->fd, iov, count);
}
static
int
send_reply_iov(fuse_req_t *req,
int error,
struct iovec *iov,
int count)
{
int res;
res = fuse_send_reply_iov_nofree(req, error, iov, count);
fuse_req_free(req);
return res;
}
static
int
send_reply(fuse_req_t *req,
int error,
const void *arg,
size_t argsize)
{
struct iovec iov[2];
int count = 1;
if(argsize)
{
iov[1].iov_base = (void *) arg;
iov[1].iov_len = argsize;
count++;
}
return send_reply_iov(req, error, iov, count);
}
static
void
convert_statfs(const struct statvfs *stbuf,
struct fuse_kstatfs *kstatfs)
{
kstatfs->bsize = stbuf->f_bsize;
kstatfs->frsize = stbuf->f_frsize;
kstatfs->blocks = stbuf->f_blocks;
kstatfs->bfree = stbuf->f_bfree;
kstatfs->bavail = stbuf->f_bavail;
kstatfs->files = stbuf->f_files;
kstatfs->ffree = stbuf->f_ffree;
kstatfs->namelen = stbuf->f_namemax;
}
static
int
send_reply_ok(fuse_req_t *req,
const void *arg,
size_t argsize)
{
return send_reply(req, 0, arg, argsize);
}
int
fuse_reply_err(fuse_req_t *req_,
int err_)
{
if(fuse_cfg.debug)
{
struct fuse_out_header hdr = {};
hdr.unique = req_->ctx.unique;
hdr.error = (err_ > 0) ? -err_ : err_;
hdr.len = sizeof(struct fuse_out_header);
fuse_debug_out_header(&hdr);
}
return send_reply(req_,err_,NULL,0);
}
void
fuse_reply_none(fuse_req_t *req)
{
fuse_req_free(req);
}
static
void
fill_entry(struct fuse_entry_out *arg,
const struct fuse_entry_param *e)
{
arg->nodeid = e->ino;
arg->generation = e->generation;
arg->entry_valid = e->timeout.entry;
arg->entry_valid_nsec = 0;
arg->attr_valid = e->timeout.attr;
arg->attr_valid_nsec = 0;
convert_stat(&e->attr,&arg->attr);
}
static
void
fill_open(struct fuse_open_out *arg_,
const fuse_file_info_t *ffi_)
{
arg_->fh = ffi_->fh;
if(ffi_->direct_io)
arg_->open_flags |= FOPEN_DIRECT_IO;
if(ffi_->keep_cache)
arg_->open_flags |= FOPEN_KEEP_CACHE;
if(ffi_->nonseekable)
arg_->open_flags |= FOPEN_NONSEEKABLE;
if(ffi_->cache_readdir)
arg_->open_flags |= FOPEN_CACHE_DIR;
if(ffi_->parallel_direct_writes)
arg_->open_flags |= FOPEN_PARALLEL_DIRECT_WRITES;
if(ffi_->noflush)
arg_->open_flags |= FOPEN_NOFLUSH;
if(ffi_->passthrough && fuse_backing_id_is_valid(ffi_->backing_id))
{
arg_->open_flags |= FOPEN_PASSTHROUGH;
arg_->backing_id = ffi_->backing_id;
}
}
int
fuse_reply_entry(fuse_req_t *req,
const struct fuse_entry_param *e)
{
struct fuse_entry_out arg = {};
size_t size = req->conn.proto_minor < 9 ?
FUSE_COMPAT_ENTRY_OUT_SIZE : sizeof(arg);
/* before ABI 7.4 e->ino == 0 was invalid, only ENOENT meant
negative entry */
if(!e->ino && req->conn.proto_minor < 4)
return fuse_reply_err(req, ENOENT);
fill_entry(&arg, e);
#ifdef NDEBUG
// TODO: Add checks for cases where a node could be marked bad by
// the kernel.
#endif
if(fuse_cfg.debug)
fuse_debug_entry_out(req->ctx.unique,&arg,size);
return send_reply_ok(req, &arg, size);
}
struct fuse_create_out
{
struct fuse_entry_out e;
struct fuse_open_out o;
};
int
fuse_reply_create(fuse_req_t *req,
const struct fuse_entry_param *e,
const fuse_file_info_t *f)
{
struct fuse_create_out buf = {};
size_t entrysize = req->conn.proto_minor < 9 ?
FUSE_COMPAT_ENTRY_OUT_SIZE : sizeof(struct fuse_entry_out);
struct fuse_entry_out *earg = (struct fuse_entry_out*)&buf.e;
struct fuse_open_out *oarg = (struct fuse_open_out*)(((char*)&buf)+entrysize);
fill_entry(earg, e);
fill_open(oarg, f);
if(fuse_cfg.debug)
fuse_debug_entry_open_out(req->ctx.unique,earg,oarg);
return send_reply_ok(req, &buf, entrysize + sizeof(struct fuse_open_out));
}
int
fuse_reply_attr(fuse_req_t *req,
const struct stat *attr,
const uint64_t timeout)
{
struct fuse_attr_out arg = {};
size_t size = req->conn.proto_minor < 9 ?
FUSE_COMPAT_ATTR_OUT_SIZE : sizeof(arg);
arg.attr_valid = timeout;
arg.attr_valid_nsec = 0;
convert_stat(attr,&arg.attr);
if(fuse_cfg.debug)
fuse_debug_attr_out(req->ctx.unique,&arg,size);
return send_reply_ok(req,&arg,size);
}
int
fuse_reply_statx(fuse_req_t *req_,
int flags_,
struct fuse_statx *st_,
uint64_t timeout_)
{
struct fuse_statx_out outarg{};
outarg.flags = flags_;
outarg.attr_valid = timeout_;
outarg.attr_valid_nsec = 0;
outarg.stat = *(struct fuse_statx*)st_;
if(fuse_cfg.debug)
fuse_debug_statx_out(req_->ctx.unique,&outarg);
return send_reply_ok(req_,&outarg,sizeof(outarg));
}
int
fuse_reply_readlink(fuse_req_t *req,
const char *linkname,
const size_t linkname_len)
{
if(not linkname)
return fuse_reply_err(req,EIO);
if(fuse_cfg.debug)
fuse_debug_readlink(req->ctx.unique,linkname,linkname_len);
return send_reply_ok(req,linkname,linkname_len);
}
int
fuse_reply_open(fuse_req_t *req,
const fuse_file_info_t *f)
{
struct fuse_open_out arg = {};
fill_open(&arg, f);
if(fuse_cfg.debug)
fuse_debug_open_out(req->ctx.unique,&arg,sizeof(arg));
return send_reply_ok(req, &arg, sizeof(arg));
}
int
fuse_reply_write(fuse_req_t *req,
size_t count)
{
struct fuse_write_out arg = {};
arg.size = count;
if(fuse_cfg.debug)
fuse_debug_write_out(req->ctx.unique,&arg);
return send_reply_ok(req, &arg, sizeof(arg));
}
int
fuse_reply_copy_file_range_64(fuse_req_t *req,
size_t count)
{
struct fuse_copy_file_range_out arg = {};
arg.bytes_copied = count;
return send_reply_ok(req, &arg, sizeof(arg));
}
int
fuse_reply_buf(fuse_req_t *req,
const char *buf,
size_t size)
{
if(fuse_cfg.debug)
fuse_debug_data_out(req->ctx.unique,size);
return send_reply_ok(req, buf, size);
}
#if defined(__linux__)
/* Automatic splice of large READ replies (branch fd -> pipe -> /dev/fuse).
The copy path costs read + pread + writev and copies the payload
twice. The splice path costs read + write(header) + splice(fd ->
pipe) + splice(pipe -> /dev/fuse) and never copies the payload into
userspace; the kernel copies it once, into the request's pages. The
caller gates on fuse_reply_splice_fits (fuse_lowlevel.h).
Every staging failure is recoverable: nothing has reached the kernel,
so the pipe is drained and the caller replies through the copy path.
Only a kernel without splice (ENOSYS), or a seccomp policy refusing
it, disables the path for the whole process. A pipe-max-size too
small for a reply only caps the reply size that is spliced
(g_reply_pipe_limit), and a branch that rejects splice is reported
per open file (FuseReplyResult::unsupported). */
static std::atomic<bool> g_reply_splice_off{false};
/* Smallest reply pipe, in bytes, found to exceed the pipe-max-size
sysctl, which caps unprivileged F_SETPIPE_SZ; a 1MiB reply plus its
header page needs one page more than the default 1MiB. Replies that
need at least this much stay on the copy path unless this thread's
pipe already holds them; smaller replies keep splicing. A refusal
from the per-user pipe quota is not recorded: it lifts as pipes are
closed. 0 = no refusal seen. */
static std::atomic<size_t> g_reply_pipe_limit{0};
/* The pipe-max-size sysctl, read once at session setup. Read lazily, on
the first refused resize, it could fail for the same shortage of
descriptors that made that resize or pipe fail and then read as 0 for
good. 0 when it cannot be read, in which case every EPERM is treated
as that limit. */
static std::atomic<size_t> g_pipe_max_size{0};
static
void
_read_pipe_max_size(void)
{
unsigned long n = 0;
FILE *f = ::fopen("/proc/sys/fs/pipe-max-size","re");
if(f == NULL)
return;
if(::fscanf(f,"%lu",&n) == 1)
g_pipe_max_size.store((size_t)n,std::memory_order_relaxed);
::fclose(f);
}
static
void
_reply_splice_disable(const char *what_,
const int err_)
{
if(g_reply_splice_off.exchange(true))
return;
SysLog::notice("read reply splice disabled: {} failed: {};"
" using copy path",
what_,strerror(err_));
}
/* One latch per distinct failure reason: these are independent,
recurring conditions (a pipe quota hiccup, a header-staging error, a
branch-splice error, a submission error, ...), so silencing all of
them after whichever happens to fire first would hide later,
unrelated failures from the log. A submission the kernel refused
before it looked at the request (the copy path then answers it) is
a different incident from one that fails the request, so they do
not share a latch. */
enum class SpliceWarnReason
{
pipe_quota,
pipe_create,
header_stage,
branch_splice,
submit_fallback,
submit_failed,
_count
};
static std::atomic<bool> g_reply_splice_warned[(size_t)SpliceWarnReason::_count] = {};
static
void
_reply_splice_warn(const char *what_,
const int err_,
const SpliceWarnReason reason_,
const char *outcome_ =
"falling back to copy path for this reply")
{
if(g_reply_splice_warned[(size_t)reason_].exchange(true))
return;
SysLog::warning("read reply splice: {} failed: {}; {}",
what_,strerror(err_),outcome_);
}
/* pipe_quota (the per-user pipe count/memory limit) is the one reason
that is genuinely transient -- it lifts as pipes are closed, unlike
the others, which are properties of the kernel or a branch fd that
do not change at runtime. Clear its latch once a resize succeeds, so
a later recurrence (a distinct incident) is logged again instead of
being silenced forever by whichever incident fired first. */
static
void
_reply_splice_warn_clear(const SpliceWarnReason reason_)
{
g_reply_splice_warned[(size_t)reason_].store(false,std::memory_order_relaxed);
}
/* Whether splice itself is unusable in this process: missing from the
kernel (ENOSYS) or refused by a seccomp policy, which commonly
answers EPERM but may use any errno. A zero-length splice returns 0
before looking at its descriptors, so only then does it fail. An
error from splicing a branch file normally comes from that file's
own read path instead (ENOSYS from a FUSE branch whose server does
not implement READ, EIO, ...), which this tells apart. */
static
bool
_splice_syscall_unusable()
{
return (::splice(-1,nullptr,-1,nullptr,0,0) == -1);
}
static
void
fuse_send_errno(const int fd_,
const int errno_,
const uint64_t unique_id_);
namespace
{
/* One pipe per worker thread, created on first use and reused for
every large reply that thread sends. Empty between replies: every
path that leaves bytes behind drains or closes it. */
struct ReplyPipe
{
int fd[2] = {-1,-1};
size_t cap = 0;
~ReplyPipe() { reset(); }
void
reset()
{
if(fd[0] != -1) ::close(fd[0]);
if(fd[1] != -1) ::close(fd[1]);
fd[0] = fd[1] = -1;
cap = 0;
}
/* 0 when a pipe of at least bytes_ capacity is held, else -errno. */
int
ready(const size_t bytes_)
{
if(fd[0] == -1)
{
/* The initial capacity is not queried: cap 0 makes the
first use resize to exactly what it needs, which with 4KiB
pages it would have to do anyway (a reply of 64KiB plus
its header needs more than the default 16 slots). */
if(::pipe2(fd,O_CLOEXEC|O_NONBLOCK) == -1)
return -errno;
}
if(bytes_ <= cap)
return 0;
if(bytes_ > (size_t)INT_MAX)
return -EINVAL;
const int c = ::fcntl(fd[0],F_SETPIPE_SZ,(int)bytes_);
if(c == -1)
return -errno;
cap = (size_t)c;
return ((cap >= bytes_) ? 0 : -ENOSPC);
}
/* /dev/null, opened on first use and kept. A failed open (EMFILE at
that moment, say) is retried by the next drain rather than
cached, which would leave every later drain copying. */
static
int
devnull()
{
static std::atomic<int> g_fd{-1};
int fd = g_fd.load(std::memory_order_acquire);
if(fd != -1)
return fd;
fd = ::open("/dev/null",O_WRONLY|O_CLOEXEC);
if(fd == -1)
return -1;
int expected = -1;
if(g_fd.compare_exchange_strong(expected,fd,std::memory_order_acq_rel))
return fd;
::close(fd);
return expected;
}
/* Discard anything staged. Splicing it into /dev/null drops the
page references without copying the data into userspace; reading
it out is the fallback when that is not available (no /dev/null,
a buffer error), so the pipe/capacity survive to be reused. Both
ends are non-blocking, so each stops at empty (EAGAIN); a pipe
that cannot be emptied by either is closed instead, and ready()
makes a new one on next use. */
void
drain()
{
/* true once the pipe reads empty (EAGAIN), false on any other
outcome. */
const auto pump = [](auto op_)
{
for(;;)
{
const ssize_t rv = op_();
if((rv > 0) || ((rv == -1) && (errno == EINTR)))
continue;
return ((rv == -1) && (errno == EAGAIN));
}
};
const int devnull = ReplyPipe::devnull();
if((devnull != -1) &&
pump([&]{ return ::splice(fd[0],nullptr,devnull,nullptr,
(size_t)1 << 30,SPLICE_F_NONBLOCK); }))
return;
char buf[16384];
if(pump([&]{ return ::read(fd[0],buf,sizeof(buf)); }))
return;
reset();
}
};
}
static thread_local ReplyPipe t_reply_pipe;
/* Pipe capacity a reply needs: one slot for the header plus the file
range, which may start mid-page. The kernel fills pipes one page per
slot, so capacity is counted in pages, not bytes. */
static
size_t
_reply_pipe_need(const off_t offset_,
const size_t size_)
{
const size_t slots =
(((uintptr_t)offset_ % pagesize) + size_ + pagesize - 1) / pagesize;
return (1 + slots) * pagesize;
}
/* Whether a pipe of need_ bytes may be used: always below any recorded
pipe-max-size refusal, and above it only when this thread's pipe
already has that capacity (so no resize is attempted). */
static
bool
_reply_pipe_allowed(const size_t need_)
{
const size_t limit = g_reply_pipe_limit.load(std::memory_order_relaxed);
return ((limit == 0) || (need_ < limit) || (need_ <= t_reply_pipe.cap));
}
/* The pipe capacity a reply of size_ bytes at offset_ needs, or 0 when
it is not spliced: splice is off, the reply is below the threshold
(cheaper through read + writev), or it needs a pipe over a recorded
pipe-max-size refusal. The one predicate behind both
fuse_reply_splice_fits and fuse_reply_data_splice_fd. */
static
size_t
_reply_splice_need(const off_t offset_,
const size_t size_)
{
if(g_reply_splice_off.load(std::memory_order_relaxed) ||
(size_ < FUSE_SPLICE_REPLY_MIN))
return 0;
const size_t need = _reply_pipe_need(offset_,size_);
return (_reply_pipe_allowed(need) ? need : 0);
}
bool
fuse_reply_splice_fits(const off_t offset_,
const size_t size_)
{
return (_reply_splice_need(offset_,size_) != 0);
}
FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *req_,
FileRegion &region_)
{
const size_t need = _reply_splice_need(region_.offset,region_.size);
if(need == 0)
return FuseReplyResult::fallback;
ReplyPipe &p = t_reply_pipe;
struct fuse_out_header out = {};
out.unique = req_->ctx.unique;
out.len = sizeof(out) + region_.size;
const size_t prev_cap = p.cap;
int err = p.ready(need);
if(err != 0)
{
/* EPERM means either larger than pipe-max-size without
CAP_SYS_RESOURCE, a fixed bound worth remembering so replies
this large skip straight to the copy path, or over the
per-user pipe quota, which is transient: only this reply falls
back. An unreadable sysctl is treated as the fixed bound. */
const size_t max = g_pipe_max_size.load(std::memory_order_relaxed);
if((err == -EPERM) && ((max == 0) || (need > max)))
{
size_t cur = g_reply_pipe_limit.load(std::memory_order_relaxed);
while(((cur == 0) || (need < cur)) &&
!g_reply_pipe_limit.compare_exchange_weak(cur,need))
;
/* cur is still 0 only in the thread that set the first limit. */
if(cur == 0)
SysLog::notice("read reply splice: a {} byte pipe exceeds"
" pipe-max-size; reads this large use the"
" copy path, smaller reads are still spliced",
need);
}
else if(err == -EPERM)
_reply_splice_warn("growing the reply pipe (per-user pipe quota)",-err,
SpliceWarnReason::pipe_quota);
else if(err == -ENOSYS)
_reply_splice_disable("creating the reply pipe",-err);
else
_reply_splice_warn("creating the reply pipe",-err,
SpliceWarnReason::pipe_create);
return FuseReplyResult::fallback;
}
/* A resize just succeeded, so any earlier per-user pipe quota
shortage (the one transient SpliceWarnReason) has lifted; let a
later recurrence be logged again instead of staying silenced by an
earlier one. A pipe that was already big enough says nothing about
the quota: it must not re-arm the warning while other threads still
fail to grow theirs, nor write the shared latch on every reply. */
if(p.cap != prev_cap)
_reply_splice_warn_clear(SpliceWarnReason::pipe_quota);
/* write(), not vmsplice(): vmsplice without SPLICE_F_GIFT queues a
reference to this stack page rather than a copy, and the pipe
outlives the frame whenever it is consumed later (or not at all).
A 16-byte write into an empty pipe copies and is atomic. */
ssize_t n;
do
n = ::write(p.fd[1],&out,sizeof(out));
while((n == -1) && (errno == EINTR));
if(n != (ssize_t)sizeof(out))
{
err = ((n == -1) ? errno : EIO);
p.drain();
_reply_splice_warn("staging the reply header",err,
SpliceWarnReason::header_stage);
return FuseReplyResult::fallback;
}
size_t done = 0;
while(done < region_.size)
{
loff_t pos = region_.offset + (off_t)done;
n = ::splice(region_.fd,&pos,p.fd[1],nullptr,
region_.size - done,SPLICE_F_NONBLOCK);
if((n == -1) && (errno == EINTR))
continue;
if(n > 0)
{
done += (size_t)n;
continue;
}
/* n == 0: the file ended before the region did (it shrank after
the caller sized the region). n == -1: this descriptor rejects
splice (EINVAL: its filesystem has no splice_read) or failed.
EAGAIN is also end-of-file when the file ended inside the
region's last page of an exactly full pipe (no room is left
for the splice that would return 0). Nothing reached the
kernel. A rejecting descriptor is a property of that branch
file, not of the transport, so the path stays on and the
caller is told to stop offering it; only a first splice can
show that, since one that already moved data proves the
descriptor splices. An error disables the path only when
splice itself is unusable (missing, or refused by seccomp),
not when it is the branch file's read error; otherwise every
large read would pay for a staging attempt that cannot
succeed. */
err = ((n == -1) ? errno : 0);
p.drain();
if((err != 0) && (err != EAGAIN))
{
if(_splice_syscall_unusable())
_reply_splice_disable("splicing from the branch file",err);
else if(err != EINVAL)
_reply_splice_warn("splicing from the branch file",err,
SpliceWarnReason::branch_splice);
}
return (((err == EINVAL) && (done == 0)) ?
FuseReplyResult::unsupported :
FuseReplyResult::fallback);
}
/* fd is not read again on any outcome below (see FileRegion::lock). */
if(region_.lock.owns_lock())
region_.lock.unlock();
/* The whole message is staged now, so a log write that throws (fmt on
a full or broken log file) must not leave it in this thread's pipe:
the worker survives the exception and would splice it in front of
its next reply. */
if(fuse_cfg.debug)
{
try
{
fuse_debug_data_out(req_->ctx.unique,region_.size);
}
catch(...)
{
p.drain();
throw;
}
}
/* Pages spliced in from the branch page cache are never offered for
SPLICE_F_MOVE: stealing them would evict the branch's cache pages.
The kernel copies them into the request's pages instead. */
const size_t total = out.len;
do
n = ::splice(p.fd[0],nullptr,req_->fd,nullptr,total,0);
while((n == -1) && (errno == EINTR));
if(n == (ssize_t)total)
return FuseReplyResult::sent;
err = ((n == -1) ? errno : EIO);
/* The kernel takes the whole message or none of it. If the pipe still
holds all of it, the submission failed before the kernel looked at
the request (e.g. ENOMEM allocating its buffer list), which is
still pending, so the copy path can answer it. */
int queued = 0;
if((::ioctl(p.fd[0],FIONREAD,&queued) == 0) && ((size_t)queued == total))
{
p.drain();
_reply_splice_warn("submitting the reply",err,
SpliceWarnReason::submit_fallback);
return FuseReplyResult::fallback;
}
/* Otherwise the kernel may have completed the request before failing
(it validates after copying; a copy error fails it with EIO), so
the payload is never replayed. ENOENT means the request was already
retired or the connection aborted: nothing is left to answer, as
when the copy path's writev fails the same way, so it is neither
logged nor answered. Any other error-only reply completes a request
that is still pending. */
p.drain();
if(err != ENOENT)
{
_reply_splice_warn("submitting the reply",err,
SpliceWarnReason::submit_failed,
"failing the request");
fuse_send_errno(req_->fd,err,req_->ctx.unique);
}
return FuseReplyResult::failed;
}
#else
bool
fuse_reply_splice_fits(const off_t,
const size_t)
{
return false;
}
FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *,
FileRegion &)
{
return FuseReplyResult::fallback;
}
#endif
int
fuse_reply_data(fuse_req_t *req,
char *buf_,
const size_t bufsize_)
{
int res;
struct iovec iov[2];
struct fuse_out_header out;
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(struct fuse_out_header);
iov[1].iov_base = buf_;
iov[1].iov_len = bufsize_;
out.unique = req->ctx.unique;
out.error = 0;
if(fuse_cfg.debug)
fuse_debug_data_out(req->ctx.unique,bufsize_);
res = fuse_send_msg(req->fd,iov,2);
fuse_req_free(req);
return res;
}
int
fuse_reply_statfs(fuse_req_t *req,
const struct statvfs *stbuf)
{
struct fuse_statfs_out arg = {};
size_t size = req->conn.proto_minor < 4 ?
FUSE_COMPAT_STATFS_SIZE : sizeof(arg);
convert_statfs(stbuf, &arg.st);
if(fuse_cfg.debug)
fuse_debug_statfs_out(req->ctx.unique,&arg);
return send_reply_ok(req, &arg, size);
}
int
fuse_reply_xattr(fuse_req_t *req,
size_t count)
{
struct fuse_getxattr_out arg = {};
arg.size = count;
if(fuse_cfg.debug)
fuse_debug_getxattr_out(req->ctx.unique,&arg);
return send_reply_ok(req, &arg, sizeof(arg));
}
int
fuse_reply_bmap(fuse_req_t *req,
uint64_t idx)
{
struct fuse_bmap_out arg = {};
arg.block = idx;
if(fuse_cfg.debug)
fuse_debug_bmap_out(req->ctx.unique,&arg);
return send_reply_ok(req, &arg, sizeof(arg));
}
static
struct fuse_ioctl_iovec*
fuse_ioctl_iovec_copy(const struct iovec *iov,
size_t count)
{
struct fuse_ioctl_iovec *fiov;
size_t i;
fiov = (fuse_ioctl_iovec*)malloc(sizeof(fiov[0]) * count);
if(!fiov)
return NULL;
for (i = 0; i < count; i++)
{
fiov[i].base = (uintptr_t) iov[i].iov_base;
fiov[i].len = iov[i].iov_len;
}
return fiov;
}
int
fuse_reply_ioctl_retry(fuse_req_t *req,
const struct iovec *in_iov,
size_t in_count,
const struct iovec *out_iov,
size_t out_count)
{
int res;
struct fuse_ioctl_out arg = {};
struct fuse_ioctl_iovec *in_fiov = NULL;
struct fuse_ioctl_iovec *out_fiov = NULL;
struct iovec iov[4];
size_t count = 1;
arg.flags |= FUSE_IOCTL_RETRY;
arg.in_iovs = in_count;
arg.out_iovs = out_count;
if(fuse_cfg.debug)
fuse_debug_ioctl_out(req->ctx.unique,&arg);
iov[count].iov_base = &arg;
iov[count].iov_len = sizeof(arg);
count++;
if(req->conn.proto_minor < 16)
{
if(in_count)
{
iov[count].iov_base = (void *)in_iov;
iov[count].iov_len = sizeof(in_iov[0]) * in_count;
count++;
}
if(out_count)
{
iov[count].iov_base = (void *)out_iov;
iov[count].iov_len = sizeof(out_iov[0]) * out_count;
count++;
}
}
else
{
/* Can't handle non-compat 64bit ioctls on 32bit */
if((sizeof(void *) == 4) && (req->ioctl_64bit))
{
res = fuse_reply_err(req, EINVAL);
goto out;
}
if(in_count)
{
in_fiov = fuse_ioctl_iovec_copy(in_iov, in_count);
if(!in_fiov)
goto enomem;
iov[count].iov_base = (void *)in_fiov;
iov[count].iov_len = sizeof(in_fiov[0]) * in_count;
count++;
}
if(out_count)
{
out_fiov = fuse_ioctl_iovec_copy(out_iov, out_count);
if(!out_fiov)
goto enomem;
iov[count].iov_base = (void *)out_fiov;
iov[count].iov_len = sizeof(out_fiov[0]) * out_count;
count++;
}
}
res = send_reply_iov(req, 0, iov, count);
out:
free(in_fiov);
free(out_fiov);
return res;
enomem:
res = fuse_reply_err(req, ENOMEM);
goto out;
}
int
fuse_reply_ioctl(fuse_req_t *req,
int result,
const void *buf,
uint32_t size)
{
int count;
struct iovec iov[3];
struct fuse_ioctl_out arg;
arg.result = result;
arg.flags = 0;
arg.in_iovs = 0;
arg.out_iovs = 0;
if(fuse_cfg.debug)
fuse_debug_ioctl_out(req->ctx.unique,&arg);
count = 1;
iov[count].iov_base = &arg;
iov[count].iov_len = sizeof(arg);
count++;
if(size)
{
iov[count].iov_base = (char*)buf;
iov[count].iov_len = size;
count++;
}
return send_reply_iov(req, 0, iov, count);
}
int
fuse_reply_poll(fuse_req_t *req,
unsigned revents)
{
struct fuse_poll_out arg = {};
arg.revents = revents;
if(fuse_cfg.debug)
fuse_debug_poll_out(req->ctx.unique,&arg);
return send_reply_ok(req, &arg, sizeof(arg));
}
static
void
do_lookup(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.lookup(req,hdr_);
}
static
void
do_forget(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.forget(req,hdr_);
}
static
void
do_batch_forget(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.forget_multi(req,hdr_);
}
static
void
do_getattr(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.getattr(req, hdr_);
}
static
void
do_setattr(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.setattr(req_,hdr_);
}
static
void
do_access(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.access(req,hdr_);
}
static
void
do_readlink(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.readlink(req,hdr_);
}
static
void
do_mknod(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.mknod(req,hdr_);
}
static
void
do_mkdir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.mkdir(req,hdr_);
}
static
void
do_unlink(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.unlink(req,hdr_);
}
static
void
do_rmdir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.rmdir(req,hdr_);
}
static
void
do_symlink(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.symlink(req,hdr_);
}
static
void
do_rename(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.rename(req,hdr_);
}
static
void
do_link(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.link(req,hdr_);
}
static
void
do_create(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.create(req,hdr_);
}
static
void
do_open(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.open(req,hdr_);
}
static
void
do_read(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.read(req,hdr_);
}
static
void
do_write(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.write(req,hdr_);
}
static
void
do_flush(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.flush(req,hdr_);
}
static
void
do_release(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.release(req,hdr_);
}
static
void
do_fsync(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.fsync(req,hdr_);
}
static
void
do_opendir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.opendir(req,hdr_);
}
static
void
do_readdir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.readdir(req,hdr_);
}
static
void
do_readdirplus(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.readdir_plus(req_,hdr_);
}
static
void
do_releasedir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.releasedir(req,hdr_);
}
static
void
do_fsyncdir(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.fsyncdir(req,hdr_);
}
static
void
do_statfs(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.statfs(req,hdr_);
}
static
void
do_setxattr(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.setxattr(req,hdr_);
}
static
void
do_getxattr(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.getxattr(req,hdr_);
}
static
void
do_listxattr(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.listxattr(req,hdr_);
}
static
void
do_removexattr(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.removexattr(req,hdr_);
}
void
do_interrupt(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
fuse_req_free(req);
}
static
void
do_bmap(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.bmap(req,hdr_);
}
static
void
do_ioctl(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.ioctl(req, hdr_);
}
void
fuse_pollhandle_destroy(fuse_pollhandle_t *ph)
{
free(ph);
}
static
void
do_poll(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.poll(req,hdr_);
}
static
void
do_fallocate(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.op.fallocate(req,hdr_);
}
static
void
do_init(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
struct fuse_init_out outarg = {};
struct fuse_init_in *arg = (struct fuse_init_in *)&hdr_[1];
size_t bufsize;
uint64_t inargflags;
uint64_t outargflags;
u32 max_write;
u32 max_readahead;
max_write = UINT_MAX;
max_readahead = UINT_MAX;
bufsize = req->se->bufsize;
inargflags = 0;
outargflags = 0;
fuse_syslog_fuse_init_in(arg);
f.conn.proto_major = arg->major;
f.conn.proto_minor = arg->minor;
f.conn.capable = 0;
f.conn.want = 0;
outarg.major = FUSE_KERNEL_VERSION;
outarg.minor = FUSE_KERNEL_MINOR_VERSION;
outarg.max_pages = FUSE_DEFAULT_MAX_PAGES_PER_REQ;
if(arg->major < 7)
{
fprintf(stderr, "fuse: unsupported protocol version: %u.%u\n",
arg->major, arg->minor);
fuse_reply_err(req, EPROTO);
return;
}
if(arg->major > 7)
{
/* Wait for a second INIT request with a 7.X version */
send_reply_ok(req, &outarg, sizeof(outarg));
return;
}
if(arg->minor >= 6)
{
inargflags = arg->flags;
if(inargflags & FUSE_INIT_EXT)
inargflags |= (((uint64_t)arg->flags2) << 32);
if(arg->max_readahead < max_readahead)
max_readahead = arg->max_readahead;
if(inargflags & FUSE_ASYNC_READ)
f.conn.capable |= FUSE_CAP_ASYNC_READ;
if(inargflags & FUSE_ATOMIC_O_TRUNC)
f.conn.capable |= FUSE_CAP_ATOMIC_O_TRUNC;
if(inargflags & FUSE_EXPORT_SUPPORT)
f.conn.capable |= FUSE_CAP_EXPORT_SUPPORT;
if(inargflags & FUSE_BIG_WRITES)
f.conn.capable |= FUSE_CAP_BIG_WRITES;
if(inargflags & FUSE_DONT_MASK)
f.conn.capable |= FUSE_CAP_DONT_MASK;
if(inargflags & FUSE_POSIX_ACL)
f.conn.capable |= FUSE_CAP_POSIX_ACL;
if(inargflags & FUSE_CACHE_SYMLINKS)
f.conn.capable |= FUSE_CAP_CACHE_SYMLINKS;
if(inargflags & FUSE_ASYNC_DIO)
f.conn.capable |= FUSE_CAP_ASYNC_DIO;
if(inargflags & FUSE_PARALLEL_DIROPS)
f.conn.capable |= FUSE_CAP_PARALLEL_DIROPS;
if(inargflags & FUSE_MAX_PAGES)
f.conn.capable |= FUSE_CAP_MAX_PAGES;
if(inargflags & FUSE_WRITEBACK_CACHE)
f.conn.capable |= FUSE_CAP_WRITEBACK_CACHE;
if(inargflags & FUSE_DO_READDIRPLUS)
f.conn.capable |= FUSE_CAP_READDIR_PLUS;
if(inargflags & FUSE_READDIRPLUS_AUTO)
f.conn.capable |= FUSE_CAP_READDIR_PLUS_AUTO;
if(inargflags & FUSE_SETXATTR_EXT)
f.conn.capable |= FUSE_CAP_SETXATTR_EXT;
if(inargflags & FUSE_DIRECT_IO_ALLOW_MMAP)
f.conn.capable |= FUSE_CAP_DIRECT_IO_ALLOW_MMAP;
if(inargflags & FUSE_CREATE_SUPP_GROUP)
f.conn.capable |= FUSE_CAP_CREATE_SUPP_GROUP;
if(inargflags & FUSE_PASSTHROUGH)
f.conn.capable |= FUSE_CAP_PASSTHROUGH;
if(inargflags & FUSE_HANDLE_KILLPRIV)
f.conn.capable |= FUSE_CAP_HANDLE_KILLPRIV;
if(inargflags & FUSE_HANDLE_KILLPRIV_V2)
f.conn.capable |= FUSE_CAP_HANDLE_KILLPRIV_V2;
if(inargflags & FUSE_ALLOW_IDMAP)
f.conn.capable |= FUSE_CAP_ALLOW_IDMAP;
}
else
{
f.conn.want &= ~FUSE_CAP_ASYNC_READ;
max_readahead = 0;
}
if(f.conn.proto_minor >= 18)
f.conn.capable |= FUSE_CAP_IOCTL_DIR;
if(bufsize < FUSE_MIN_READ_BUFFER)
{
fprintf(stderr, "fuse: warning: buffer size too small: %zu\n",
bufsize);
bufsize = FUSE_MIN_READ_BUFFER;
}
bufsize -= pagesize;
if(bufsize < max_write)
max_write = bufsize;
f.got_init = 1;
f.op.init(&f.conn);
outargflags = outarg.flags;
if((inargflags & FUSE_MAX_PAGES) && (f.conn.want & FUSE_CAP_MAX_PAGES))
{
outargflags |= FUSE_MAX_PAGES;
outarg.max_pages = fuse_cfg.max_pages;
msgbuf_set_bufsize(outarg.max_pages);
max_write = (msgbuf_get_pagesize() * outarg.max_pages);
}
if(f.conn.want & FUSE_CAP_ASYNC_READ)
outargflags |= FUSE_ASYNC_READ;
if(f.conn.want & FUSE_CAP_ATOMIC_O_TRUNC)
outargflags |= FUSE_ATOMIC_O_TRUNC;
if(f.conn.want & FUSE_CAP_EXPORT_SUPPORT)
outargflags |= FUSE_EXPORT_SUPPORT;
if(f.conn.want & FUSE_CAP_BIG_WRITES)
outargflags |= FUSE_BIG_WRITES;
if(f.conn.want & FUSE_CAP_DONT_MASK)
outargflags |= FUSE_DONT_MASK;
if(f.conn.want & FUSE_CAP_POSIX_ACL)
outargflags |= FUSE_POSIX_ACL;
if(f.conn.want & FUSE_CAP_CACHE_SYMLINKS)
outargflags |= FUSE_CACHE_SYMLINKS;
if(f.conn.want & FUSE_CAP_ASYNC_DIO)
outargflags |= FUSE_ASYNC_DIO;
if(f.conn.want & FUSE_CAP_PARALLEL_DIROPS)
outargflags |= FUSE_PARALLEL_DIROPS;
if(f.conn.want & FUSE_CAP_WRITEBACK_CACHE)
outargflags |= FUSE_WRITEBACK_CACHE;
if(f.conn.want & FUSE_CAP_READDIR_PLUS)
outargflags |= FUSE_DO_READDIRPLUS;
if(f.conn.want & FUSE_CAP_READDIR_PLUS_AUTO)
outargflags |= FUSE_READDIRPLUS_AUTO;
if(f.conn.want & FUSE_CAP_SETXATTR_EXT)
outargflags |= FUSE_SETXATTR_EXT;
if(f.conn.want & FUSE_CAP_CREATE_SUPP_GROUP)
outargflags |= FUSE_CREATE_SUPP_GROUP;
if(f.conn.want & FUSE_CAP_DIRECT_IO_ALLOW_MMAP)
outargflags |= FUSE_DIRECT_IO_ALLOW_MMAP;
if(f.conn.want & FUSE_CAP_HANDLE_KILLPRIV)
outargflags |= FUSE_HANDLE_KILLPRIV;
if(f.conn.want & FUSE_CAP_HANDLE_KILLPRIV_V2)
outargflags |= FUSE_HANDLE_KILLPRIV_V2;
if(f.conn.want & FUSE_CAP_ALLOW_IDMAP)
outargflags |= FUSE_ALLOW_IDMAP;
if(f.conn.want & FUSE_CAP_PASSTHROUGH)
{
outargflags |= FUSE_PASSTHROUGH;
outarg.max_stack_depth = fuse_cfg.passthrough_max_stack_depth;
}
if((inargflags & FUSE_REQUEST_TIMEOUT) && fuse_cfg.request_timeout)
{
outargflags |= FUSE_REQUEST_TIMEOUT;
outarg.request_timeout = fuse_cfg.request_timeout;
}
if(inargflags & FUSE_INIT_EXT)
{
outargflags |= FUSE_INIT_EXT;
outarg.flags2 = (outargflags >> 32);
}
outarg.flags = outargflags;
outarg.max_readahead = max_readahead;
outarg.max_write = max_write;
if(f.conn.proto_minor >= 13)
{
if(fuse_cfg.max_background >= (1 << 16))
fuse_cfg.max_background = ((1 << 16) - 1);
if(fuse_cfg.congestion_threshold > fuse_cfg.max_background)
fuse_cfg.congestion_threshold = fuse_cfg.max_background;
if(!fuse_cfg.congestion_threshold)
fuse_cfg.congestion_threshold = (fuse_cfg.max_background * 3 / 4);
outarg.max_background = fuse_cfg.max_background;
outarg.congestion_threshold = fuse_cfg.congestion_threshold;
}
if(f.conn.proto_minor >= 23)
outarg.time_gran = 1;
size_t outargsize;
if(arg->minor < 5)
outargsize = FUSE_COMPAT_INIT_OUT_SIZE;
else if(arg->minor < 23)
outargsize = FUSE_COMPAT_22_INIT_OUT_SIZE;
else
outargsize = sizeof(outarg);
fuse_syslog_fuse_init_out(&outarg);
if(fuse_cfg.debug)
fuse_debug_init_out(req->ctx.unique,&outarg,outargsize);
send_reply_ok(req, &outarg, outargsize);
}
static
void
do_destroy(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
f.got_destroy = 1;
f.op.destroy();
send_reply_ok(req,NULL,0);
}
static
void
list_del_nreq(struct fuse_notify_req *nreq)
{
struct fuse_notify_req *prev = nreq->prev;
struct fuse_notify_req *next = nreq->next;
prev->next = next;
next->prev = prev;
}
static
void
list_add_nreq(struct fuse_notify_req *nreq,
struct fuse_notify_req *next)
{
struct fuse_notify_req *prev = next->prev;
nreq->next = next;
nreq->prev = prev;
prev->next = nreq;
next->prev = nreq;
}
static
void
list_init_nreq(struct fuse_notify_req *nreq)
{
nreq->next = nreq;
nreq->prev = nreq;
}
static
void
do_notify_reply(fuse_req_t *req,
struct fuse_in_header *hdr_)
{
struct fuse_notify_req *nreq;
struct fuse_notify_req *head;
mutex_lock(f.lock);
head = &f.notify_list;
for(nreq = head->next; nreq != head; nreq = nreq->next)
{
if(nreq->unique == req->ctx.unique)
{
list_del_nreq(nreq);
break;
}
}
mutex_unlock(f.lock);
if(nreq != head)
nreq->reply(nreq, req, hdr_->nodeid, &hdr_[1]);
}
static
void
do_copy_file_range(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.copy_file_range(req_,hdr_);
}
static
void
do_copy_file_range_64(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.copy_file_range(req_,hdr_);
}
static
void
do_setupmapping(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.setupmapping(req_,hdr_);
}
static
void
do_removemapping(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.removemapping(req_,hdr_);
}
static
void
do_syncfs(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.syncfs(req_,hdr_);
}
static
void
do_tmpfile(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.tmpfile(req_,hdr_);
}
static
void
do_statx(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.statx(req_,hdr_);
}
static
void
do_rename2(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.rename2(req_,hdr_);
}
static
void
do_lseek(fuse_req_t *req_,
struct fuse_in_header *hdr_)
{
f.op.lseek(req_,hdr_);
}
static
int
send_notify_iov(struct fuse_session *se,
int notify_code,
struct iovec *iov,
int count)
{
struct fuse_out_header out;
if(!f.got_init)
return -ENOTCONN;
out.unique = 0;
out.error = notify_code;
iov[0].iov_base = &out;
iov[0].iov_len = sizeof(struct fuse_out_header);
return fuse_send_msg(se->fd, iov, count);
}
int
fuse_lowlevel_notify_poll(fuse_pollhandle_t *ph)
{
if(ph != NULL)
{
struct fuse_notify_poll_wakeup_out outarg;
struct iovec iov[2];
outarg.kh = ph->kh;
iov[1].iov_base = &outarg;
iov[1].iov_len = sizeof(outarg);
return send_notify_iov(ph->se, FUSE_NOTIFY_POLL, iov, 2);
}
else
{
return 0;
}
}
int
fuse_lowlevel_notify_inval_inode(struct fuse_session *se,
uint64_t ino,
off_t off,
off_t len)
{
struct fuse_notify_inval_inode_out outarg;
struct iovec iov[2];
if(!se)
return -EINVAL;
outarg.ino = ino;
outarg.off = off;
outarg.len = len;
iov[1].iov_base = &outarg;
iov[1].iov_len = sizeof(outarg);
return send_notify_iov(se, FUSE_NOTIFY_INVAL_INODE, iov, 2);
}
int
fuse_lowlevel_notify_inval_entry(struct fuse_session *se,
uint64_t parent,
const char *name,
size_t namelen)
{
struct fuse_notify_inval_entry_out outarg;
struct iovec iov[3];
if(!se)
return -EINVAL;
outarg.parent = parent;
outarg.namelen = namelen;
// TODO: Add ability to set `flags`
outarg.flags = 0;
iov[1].iov_base = &outarg;
iov[1].iov_len = sizeof(outarg);
iov[2].iov_base = (void *)name;
iov[2].iov_len = namelen + 1;
return send_notify_iov(se, FUSE_NOTIFY_INVAL_ENTRY, iov, 3);
}
int
fuse_lowlevel_notify_delete(struct fuse_session *se,
uint64_t parent,
uint64_t child,
const char *name,
size_t namelen)
{
struct fuse_notify_delete_out outarg;
struct iovec iov[3];
if(!se)
return -EINVAL;
if(f.conn.proto_minor < 18)
return -ENOSYS;
outarg.parent = parent;
outarg.child = child;
outarg.namelen = namelen;
outarg.padding = 0;
iov[1].iov_base = &outarg;
iov[1].iov_len = sizeof(outarg);
iov[2].iov_base = (void *)name;
iov[2].iov_len = namelen + 1;
return send_notify_iov(se, FUSE_NOTIFY_DELETE, iov, 3);
}
struct fuse_retrieve_req
{
struct fuse_notify_req nreq;
void *cookie;
};
static
void
fuse_ll_retrieve_reply(struct fuse_notify_req *nreq,
fuse_req_t *req,
uint64_t ino,
const void *inarg)
{
struct fuse_retrieve_req *rreq =
container_of(nreq, struct fuse_retrieve_req, nreq);
fuse_reply_none(req);
free(rreq);
}
int
fuse_lowlevel_notify_retrieve(struct fuse_session *se,
uint64_t ino,
size_t size,
off_t offset,
void *cookie)
{
struct fuse_notify_retrieve_out outarg;
struct iovec iov[2];
struct fuse_retrieve_req *rreq;
int err;
if(!se)
return -EINVAL;
if(f.conn.proto_minor < 15)
return -ENOSYS;
rreq = (fuse_retrieve_req*)malloc(sizeof(*rreq));
if(rreq == NULL)
return -ENOMEM;
mutex_lock(f.lock);
rreq->cookie = cookie;
rreq->nreq.unique = f.notify_ctr++;
rreq->nreq.reply = fuse_ll_retrieve_reply;
list_add_nreq(&rreq->nreq, &f.notify_list);
mutex_unlock(f.lock);
outarg.notify_unique = rreq->nreq.unique;
outarg.nodeid = ino;
outarg.offset = offset;
outarg.size = size;
iov[1].iov_base = &outarg;
iov[1].iov_len = sizeof(outarg);
err = send_notify_iov(se, FUSE_NOTIFY_RETRIEVE, iov, 2);
if(err)
{
mutex_lock(f.lock);
list_del_nreq(&rreq->nreq);
mutex_unlock(f.lock);
free(rreq);
}
return err;
}
#define FUSE_OPCODE_LEN (FUSE_COPY_FILE_RANGE_64 + 1)
typedef void (*fuse_ll_func)(fuse_req_t*, struct fuse_in_header *);
const
fuse_ll_func
fuse_ll_funcs[FUSE_OPCODE_LEN] =
{
NULL,
do_lookup,
do_forget,
do_getattr,
do_setattr,
do_readlink,
do_symlink,
NULL,
do_mknod,
do_mkdir,
do_unlink,
do_rmdir,
do_rename,
do_link,
do_open,
do_read,
do_write,
do_statfs,
do_release,
NULL,
do_fsync,
do_setxattr,
do_getxattr,
do_listxattr,
do_removexattr,
do_flush,
do_init,
do_opendir,
do_readdir,
do_releasedir,
do_fsyncdir,
NULL,
NULL,
NULL,
do_access,
do_create,
do_interrupt,
do_bmap,
do_destroy,
do_ioctl,
do_poll,
do_notify_reply,
do_batch_forget,
do_fallocate,
do_readdirplus,
do_rename2,
do_lseek,
do_copy_file_range,
do_setupmapping,
do_removemapping,
do_syncfs,
do_tmpfile,
do_statx,
do_copy_file_range_64
};
#define FUSE_MAXOPS (sizeof(fuse_ll_funcs) / sizeof(fuse_ll_funcs[0]))
static
void
fuse_ll_destroy(void *data)
{
if(f.got_init && !f.got_destroy)
f.op.destroy();
mutex_destroy(f.lock);
msgbuf_gc();
}
static
void
fuse_send_errno(const int fd_,
const int errno_,
const uint64_t unique_id_)
{
struct fuse_out_header out = {};
struct iovec iov = {};
out.unique = unique_id_;
out.error = -errno_;
out.len = sizeof(struct fuse_out_header);
iov.iov_base = &out;
iov.iov_len = sizeof(struct fuse_out_header);
if(fuse_cfg.debug)
fuse_debug_out_header(&out);
fuse_send_msg(fd_,&iov,1);
}
static
void
fuse_send_enomem(const int fd_,
const uint64_t unique_id_)
{
fuse_send_errno(fd_,ENOMEM,unique_id_);
}
static
int
fuse_ll_buf_receive_read(struct fuse_session *se_,
int fd_,
fuse_msgbuf_t *msgbuf_)
{
int rv;
rv = read(fd_,msgbuf_->mem,msgbuf_->size);
if(rv == -1)
return -errno;
if(rv < (int)sizeof(struct fuse_in_header))
{
fprintf(stderr, "short read from fuse device\n");
return -EIO;
}
return rv;
}
static
void
fuse_ll_buf_process_read(struct fuse_session *se_,
int fd_,
const fuse_msgbuf_t *msgbuf_)
{
int err;
struct fuse_req_t *req;
struct fuse_in_header *in;
in = (struct fuse_in_header*)msgbuf_->mem;
if(fuse_cfg.debug)
fuse_debug_in_header(in);
req = fuse_req_alloc();
if(req == NULL)
return fuse_send_enomem(fd_,in->unique);
req->ctx.len = in->len;
req->ctx.opcode = in->opcode;
req->ctx.unique = in->unique;
req->ctx.nodeid = in->nodeid;
req->ctx.uid = in->uid;
req->ctx.gid = in->gid;
req->ctx.pid = in->pid;
req->ctx.umask = 0;
req->conn = f.conn;
req->se = se_;
req->fd = fd_;
req->ioctl_64bit = 0;
err = ENOSYS;
if(in->opcode >= FUSE_MAXOPS)
goto reply_err;
if(fuse_ll_funcs[in->opcode] == NULL)
goto reply_err;
fuse_ll_funcs[in->opcode](req, in);
return;
reply_err:
fuse_reply_err(req, err);
return;
}
static
void
fuse_ll_buf_process_read_init(struct fuse_session *se_,
int fd_,
const fuse_msgbuf_t *msgbuf_)
{
int err;
fuse_req_t *req;
struct fuse_in_header *in;
in = (struct fuse_in_header*)msgbuf_->mem;
if(fuse_cfg.debug)
fuse_debug_in_header(in);
req = fuse_req_alloc();
if(req == NULL)
return fuse_send_enomem(fd_,in->unique);
req->ctx.len = in->len;
req->ctx.opcode = in->opcode;
req->ctx.unique = in->unique;
req->ctx.nodeid = in->nodeid;
req->ioctl_64bit = 0;
req->ctx.uid = in->uid;
req->ctx.gid = in->gid;
req->ctx.pid = in->pid;
req->ctx.umask = 0;
req->se = se_;
req->fd = fd_;
err = EIO;
if(in->opcode != FUSE_INIT)
goto reply_err;
if(fuse_ll_funcs[FUSE_INIT] == NULL)
goto reply_err;
se_->process_buf = fuse_ll_buf_process_read;
fuse_ll_funcs[in->opcode](req,in);
return;
reply_err:
fuse_reply_err(req, err);
return;
}
/*
* always call fuse_lowlevel_new_common() internally, to work around a
* misfeature in the FreeBSD runtime linker, which links the old
* version of a symbol to internal references.
*/
struct fuse_session *
fuse_lowlevel_new_common(struct fuse_args *args,
const struct fuse_lowlevel_ops *op,
size_t op_size)
{
struct fuse_session *se;
if(sizeof(struct fuse_lowlevel_ops) < op_size)
{
fprintf(stderr, "fuse: warning: library too old, some operations may not work\n");
op_size = sizeof(struct fuse_lowlevel_ops);
}
list_init_nreq(&f.notify_list);
f.notify_ctr = 1;
mutex_init(f.lock);
memcpy(&f.op,op,op_size);
f.owner = getuid();
#if defined(__linux__)
_read_pipe_max_size();
#endif
se = fuse_session_new(&f,
(void*)fuse_ll_buf_receive_read,
(void*)fuse_ll_buf_process_read_init,
(void*)fuse_ll_destroy);
if(!se)
goto out_free;
return se;
out_free:
mutex_destroy(f.lock);
return nullptr;
}
struct fuse_session*
fuse_lowlevel_new(struct fuse_args *args,
const struct fuse_lowlevel_ops *op,
size_t op_size)
{
return fuse_lowlevel_new_common(args, op, op_size);
}
Reference in a new issue View git blame Copy permalink