From 7ec6f8e2f2cd2b5cc56b203fb389fc1b9cc27e83 Mon Sep 17 00:00:00 2001 From: Antonio SJ Musumeci Date: Thu, 14 May 2026 07:53:15 -0500 Subject: [PATCH] Manage cloned /dev/fuse fds in sessions - Store cloned read-worker fds on fuse_session and close them there. - Set up clone fallback through session helpers used by the loop. - Carry per-request fds through receive, process, and reply paths. Tests show significant performance increases in highly concurrent workloads on a high core count system. Upto 15% better on reads and 27% on writes. --- vendored/libfuse/include/fuse_req.hpp | 1 + vendored/libfuse/lib/fuse_i.hpp | 15 +++- vendored/libfuse/lib/fuse_loop.cpp | 48 +++++++++---- vendored/libfuse/lib/fuse_lowlevel.cpp | 33 +++++---- vendored/libfuse/lib/fuse_session.cpp | 98 +++++++++++++++++++++++++- 5 files changed, 162 insertions(+), 33 deletions(-) diff --git a/vendored/libfuse/include/fuse_req.hpp b/vendored/libfuse/include/fuse_req.hpp index 7ba7df4c..e3d04d0c 100644 --- a/vendored/libfuse/include/fuse_req.hpp +++ b/vendored/libfuse/include/fuse_req.hpp @@ -10,6 +10,7 @@ struct fuse_req_t { fuse_req_ctx_t ctx; struct fuse_session *se; + int fd; fuse_conn_info_t conn; unsigned int ioctl_64bit : 1; }; diff --git a/vendored/libfuse/lib/fuse_i.hpp b/vendored/libfuse/lib/fuse_i.hpp index 8989aa07..fb395416 100644 --- a/vendored/libfuse/lib/fuse_i.hpp +++ b/vendored/libfuse/lib/fuse_i.hpp @@ -15,6 +15,7 @@ #include "extern_c.h" #include +#include /* Simplified fuse_session - fields inlined from former fuse_chan * Since mergerfs only supports one mount, we collapse the hierarchy: @@ -24,9 +25,11 @@ struct fuse_session { int (*receive_buf)(struct fuse_session *se, + int fd, fuse_msgbuf_t *msgbuf); void (*process_buf)(struct fuse_session *se, + int fd, const fuse_msgbuf_t *msgbuf); void (*destroy)(void *data); @@ -35,8 +38,10 @@ struct fuse_session std::atomic exited; /* Formerly in fuse_chan - inlined for single-mount simplicity */ - int fd; /* /dev/fuse file descriptor */ - size_t bufsize; /* Buffer size for I/O operations */ + int fd; /* /dev/fuse file descriptor */ + size_t bufsize; /* Buffer size for I/O operations */ + + std::vector clone_fds; /* cloned /dev/fuse fds for read workers */ }; struct fuse_notify_req @@ -50,6 +55,12 @@ struct fuse_notify_req struct fuse_notify_req *prev; }; +bool fuse_session_setup_read_fds(struct fuse_session *se, + int read_thread_count); + +int fuse_session_read_fd(struct fuse_session *se, + int index); + EXTERN_C_BEGIN diff --git a/vendored/libfuse/lib/fuse_loop.cpp b/vendored/libfuse/lib/fuse_loop.cpp index 408853b9..e6b3eec4 100644 --- a/vendored/libfuse/lib/fuse_loop.cpp +++ b/vendored/libfuse/lib/fuse_loop.cpp @@ -11,7 +11,6 @@ #include "maintenance_thread.hpp" #include "fuse_i.hpp" -#include "fuse_kernel.h" #include "fuse_lowlevel.h" #include "fuse_cfg.hpp" @@ -67,13 +66,16 @@ _nproc() struct AsyncWorker { fuse_session *_se; + int _fd; sem_t *_finished; std::shared_ptr _process_tp; AsyncWorker(fuse_session *se_, + int fd_, sem_t *finished_, std::shared_ptr process_tp_) : _se(se_), + _fd(fd_), _finished(finished_), _process_tp(process_tp_) { @@ -102,7 +104,7 @@ struct AsyncWorker while(true) { pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL); - rv = _se->receive_buf(_se,msgbuf); + rv = _se->receive_buf(_se,_fd,msgbuf); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL); if(rv > 0) @@ -118,9 +120,9 @@ struct AsyncWorker } _process_tp->enqueue_work(ptok, - [=]() + [se = _se,fd = _fd,msgbuf]() { - _se->process_buf(_se,msgbuf); + se->process_buf(se,fd,msgbuf); msgbuf_free(msgbuf); }); } @@ -130,12 +132,15 @@ struct AsyncWorker struct SyncWorker { fuse_session *_se; + int _fd; sem_t *_finished; SyncWorker(fuse_session *se_, + int fd_, sem_t *finished_) : _se(se_), + _fd(fd_), _finished(finished_) { } @@ -162,7 +167,7 @@ struct SyncWorker while(true) { pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL); - rv = _se->receive_buf(_se,msgbuf); + rv = _se->receive_buf(_se,_fd,msgbuf); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL); if(rv > 0) break; @@ -176,7 +181,7 @@ struct SyncWorker return ::_print_error(rv); } - _se->process_buf(_se,msgbuf); + _se->process_buf(_se,_fd,msgbuf); msgbuf_free(msgbuf); } @@ -255,15 +260,18 @@ fuse_session_loop_mt(struct fuse_session *se_, const std::string pin_threads_type_) { sem_t finished; + bool cloned_read_fds; int read_thread_count; int process_thread_count; int process_thread_queue_depth; std::vector read_threads; std::vector process_threads; - std::unique_ptr read_tp; - std::shared_ptr process_tp; sem_init(&finished,0,0); + DEFER{ sem_destroy(&finished); }; + + std::unique_ptr read_tp; + std::shared_ptr process_tp; read_thread_count = raw_read_thread_count_; process_thread_count = raw_process_thread_count_; @@ -274,21 +282,28 @@ fuse_session_loop_mt(struct fuse_session *se_, if(process_thread_count > 0) process_tp = std::make_shared(process_thread_count, - process_thread_queue_depth, - "fuse.process"); + process_thread_queue_depth, + "fuse.process"); + + cloned_read_fds = fuse_session_setup_read_fds(se_,read_thread_count); read_tp = std::make_unique(read_thread_count, - read_thread_count, - "fuse.read"); + read_thread_count, + "fuse.read"); if(process_tp) { for(auto i = 0; i < read_thread_count; i++) - read_tp->enqueue_work(AsyncWorker(se_,&finished,process_tp)); + read_tp->enqueue_work(AsyncWorker(se_, + fuse_session_read_fd(se_,i), + &finished, + process_tp)); } else { for(auto i = 0; i < read_thread_count; i++) - read_tp->enqueue_work(SyncWorker(se_,&finished)); + read_tp->enqueue_work(SyncWorker(se_, + fuse_session_read_fd(se_,i), + &finished)); } if(read_tp) @@ -301,10 +316,12 @@ fuse_session_loop_mt(struct fuse_session *se_, SysLog::info("read-thread-count={}; " "process-thread-count={}; " "process-thread-queue-depth={}; " + "fuse-dev-ioc-clone={}; " "pin-threads={};", read_thread_count, process_thread_count, process_thread_queue_depth, + cloned_read_fds, pin_threads_type_); while(!fuse_session_exited(se_)) @@ -313,7 +330,8 @@ fuse_session_loop_mt(struct fuse_session *se_, for(auto t : read_threads) pthread_cancel(t); - sem_destroy(&finished); + read_tp.reset(); + process_tp.reset(); return 0; } diff --git a/vendored/libfuse/lib/fuse_lowlevel.cpp b/vendored/libfuse/lib/fuse_lowlevel.cpp index cf79fd9a..d754fe85 100644 --- a/vendored/libfuse/lib/fuse_lowlevel.cpp +++ b/vendored/libfuse/lib/fuse_lowlevel.cpp @@ -103,16 +103,16 @@ iov_length(const struct iovec *iov, static int -fuse_send_msg(struct fuse_session *se, - struct iovec *iov, - int count) +fuse_send_msg(const int fd_, + struct iovec *iov, + int count) { int rv; struct fuse_out_header *out = (fuse_out_header*)iov[0].iov_base; out->len = iov_length(iov, count); - rv = writev(se->fd,iov,count); + rv = writev(fd_,iov,count); if(rv == -1) return -errno; @@ -144,7 +144,7 @@ fuse_send_reply_iov_nofree(fuse_req_t *req, iov[0].iov_base = &out; iov[0].iov_len = sizeof(struct fuse_out_header); - return fuse_send_msg(req->se, iov, count); + return fuse_send_msg(req->fd, iov, count); } static @@ -440,7 +440,7 @@ fuse_reply_data(fuse_req_t *req, if(fuse_cfg.debug) fuse_debug_data_out(req->ctx.unique,bufsize_); - res = fuse_send_msg(req->se,iov,2); + res = fuse_send_msg(req->fd,iov,2); fuse_req_free(req); return res; @@ -1310,7 +1310,7 @@ send_notify_iov(struct fuse_session *se, iov[0].iov_base = &out; iov[0].iov_len = sizeof(struct fuse_out_header); - return fuse_send_msg(se, iov, count); + return fuse_send_msg(se->fd, iov, count); } int @@ -1558,7 +1558,7 @@ fuse_ll_destroy(void *data) static void -fuse_send_errno(struct fuse_session *se_, +fuse_send_errno(const int fd_, const int errno_, const uint64_t unique_id_) { @@ -1574,25 +1574,26 @@ fuse_send_errno(struct fuse_session *se_, if(fuse_cfg.debug) fuse_debug_out_header(&out); - fuse_send_msg(se_,&iov,1); + fuse_send_msg(fd_,&iov,1); } static void -fuse_send_enomem(struct fuse_session *se_, +fuse_send_enomem(const int fd_, const uint64_t unique_id_) { - fuse_send_errno(se_,ENOMEM,unique_id_); + fuse_send_errno(fd_,ENOMEM,unique_id_); } static int fuse_ll_buf_receive_read(struct fuse_session *se_, + int fd_, fuse_msgbuf_t *msgbuf_) { int rv; - rv = read(se_->fd,msgbuf_->mem,msgbuf_->size); + rv = read(fd_,msgbuf_->mem,msgbuf_->size); if(rv == -1) return -errno; @@ -1608,6 +1609,7 @@ fuse_ll_buf_receive_read(struct fuse_session *se_, static void fuse_ll_buf_process_read(struct fuse_session *se_, + int fd_, const fuse_msgbuf_t *msgbuf_) { int err; @@ -1621,7 +1623,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_, req = fuse_req_alloc(); if(req == NULL) - return fuse_send_enomem(se_,in->unique); + return fuse_send_enomem(fd_,in->unique); req->ctx.len = in->len; req->ctx.opcode = in->opcode; @@ -1633,6 +1635,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_, req->ctx.umask = 0; req->conn = f.conn; req->se = se_; + req->fd = fd_; req->ioctl_64bit = 0; err = ENOSYS; @@ -1653,6 +1656,7 @@ fuse_ll_buf_process_read(struct fuse_session *se_, static void fuse_ll_buf_process_read_init(struct fuse_session *se_, + int fd_, const fuse_msgbuf_t *msgbuf_) { int err; @@ -1666,7 +1670,7 @@ fuse_ll_buf_process_read_init(struct fuse_session *se_, req = fuse_req_alloc(); if(req == NULL) - return fuse_send_enomem(se_,in->unique); + return fuse_send_enomem(fd_,in->unique); req->ctx.len = in->len; req->ctx.opcode = in->opcode; @@ -1678,6 +1682,7 @@ fuse_ll_buf_process_read_init(struct fuse_session *se_, req->ctx.pid = in->pid; req->ctx.umask = 0; req->se = se_; + req->fd = fd_; err = EIO; if(in->opcode != FUSE_INIT) diff --git a/vendored/libfuse/lib/fuse_session.cpp b/vendored/libfuse/lib/fuse_session.cpp index 4e8f682c..e63db49d 100644 --- a/vendored/libfuse/lib/fuse_session.cpp +++ b/vendored/libfuse/lib/fuse_session.cpp @@ -8,16 +8,24 @@ #include "fuse_i.hpp" #include "fuse_kernel.h" +#include "syslog.hpp" #include #include +#include #include #include #include +#include #include +#include #include #include +#ifndef O_CLOEXEC +#define O_CLOEXEC 0 +#endif + /* * Simplified session management for single-mount filesystems. * The fuse_chan structure has been removed and its fields inlined @@ -25,6 +33,42 @@ * one mount point per process. */ +static +int +_clone_fuse_fd(const int fd_) +{ + int rv; + int clone_fd; + uint32_t source_fd; + + clone_fd = ::open("/dev/fuse",O_RDWR|O_CLOEXEC); + if(clone_fd == -1) + return -errno; + + source_fd = fd_; + rv = ::ioctl(clone_fd,FUSE_DEV_IOC_CLONE,&source_fd); + if(rv == -1) + { + int err = errno; + ::close(clone_fd); + return -err; + } + + return clone_fd; +} + +static +void +_close_clone_fds(struct fuse_session *se_) +{ + for(auto fd : se_->clone_fds) + { + if(fd != -1) + ::close(fd); + } + se_->clone_fds.clear(); +} + struct fuse_session* fuse_session_new(void *data, void *receive_buf, @@ -40,8 +84,8 @@ fuse_session_new(void *data, } se->f = static_cast(data); - se->receive_buf = reinterpret_cast(receive_buf); - se->process_buf = reinterpret_cast(process_buf); + se->receive_buf = reinterpret_cast(receive_buf); + se->process_buf = reinterpret_cast(process_buf); se->destroy = reinterpret_cast(destroy); se->fd = -1; @@ -57,6 +101,8 @@ fuse_session_destroy(struct fuse_session *se) if(se->destroy) se->destroy(se->f); + ::_close_clone_fds(se); + if(se->fd != -1) { close(se->fd); @@ -93,6 +139,8 @@ fuse_session_data(struct fuse_session *se) int fuse_session_clearfd(struct fuse_session *se) { + ::_close_clone_fds(se); + int fd = se->fd; se->fd = -1; return fd; @@ -102,9 +150,55 @@ void fuse_session_setfd(struct fuse_session *se, int fd) { + ::_close_clone_fds(se); + se->fd = fd; } +bool +fuse_session_setup_read_fds(struct fuse_session *se_, + const int read_thread_count_) +{ + ::_close_clone_fds(se_); + + if(read_thread_count_ <= 1) + return false; + + se_->clone_fds.reserve(read_thread_count_ - 1); + + for(auto i = 1; i < read_thread_count_; i++) + { + int fd; + + fd = ::_clone_fuse_fd(se_->fd); + if(fd < 0) + { + int err = -fd; + + ::_close_clone_fds(se_); + SysLog::warning("failed to clone /dev/fuse fd for read threads; " + "using shared fd: {} ({})", + strerror(err), + err); + return false; + } + + se_->clone_fds.push_back(fd); + } + + return true; +} + +int +fuse_session_read_fd(struct fuse_session *se_, + const int index_) +{ + if((index_ > 0) && (index_ <= (int)se_->clone_fds.size())) + return se_->clone_fds[index_ - 1]; + + return se_->fd; +} + void fuse_session_setbufsize(struct fuse_session *se, size_t bufsize)