rt/mergerfs
1
0
Fork 0
mirror of https://github.com/trapexit/mergerfs.git synced 2026-10-25 18:40:18 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

qos: process-classified I/O priority and rate limiting

Browse source 
Adds a QoS engine that attributes each read/write to the calling
process, matches it against a ruleset, and applies that class's ioprio
and nice value to the worker thread for the duration of the request.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Grioghar 2026-09-09 16:34:55 -05:00
parent 515b0549ef
commit 34e57dbb8a
13 changed files with 1537 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

7
src/config.cpp
View file

@ -117,6 +117,7 @@ Config::Config()
process_thread_count(fuse_cfg.process_thread_count),
process_thread_queue_depth(fuse_cfg.process_thread_queue_depth),
proxy_ioprio(false),
qos(false),
read_thread_count(fuse_cfg.read_thread_count),
readahead(0),
readdir("seq"),
@ -281,6 +282,12 @@ Config::Config()
_map["readahead"] = &readahead;
_map["remember"] = &_remember;
_map["remember-nodes"] = &_remember_nodes;
_map["qos"] = &qos;
_map["qos.max-sleep-ms"] = &qos_max_sleep_ms;
_map["qos.max-sleepers"] = &qos_max_sleepers;
_map["qos.rules"] = &qos_rules;
_map["qos.ruleset"] = &qos_ruleset;
_map["qos.stats"] = &qos_stats;
_map["rename-exdev"] = &rename_exdev;
_map["scheduling-priority"] = &scheduling_priority;
_map["security-capability"] = &security_capability;

7
src/config.hpp
View file

@ -35,6 +35,7 @@
#include "config_passthrough_io.hpp"
#include "config_pid.hpp"
#include "config_proxy_ioprio.hpp"
#include "config_qos.hpp"
#include "config_rename_exdev.hpp"
#include "config_set.hpp"
#include "config_statfs.hpp"
@ -154,6 +155,12 @@ public:
TFSRef<int> process_thread_count;
TFSRef<int> process_thread_queue_depth;
ProxyIOPrio proxy_ioprio;
QoS qos;
QoSRules qos_rules;
QoSRuleSet qos_ruleset;
QoSStats qos_stats;
QoSMaxSleepers qos_max_sleepers;
QoSMaxSleepMS qos_max_sleep_ms;
TFSRef<int> read_thread_count;
ConfigU64 readahead;
FUSE::ReadDir readdir;

164
src/config_qos.cpp Normal file
View file

@ -0,0 +1,164 @@
/*
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 "config_qos.hpp"
#include "from_string.hpp"
#include "qos.hpp"
#include "syslog.hpp"
#include <errno.h>
QoS::QoS(const bool b_)
{
qos::enable(b_);
}
std::string
QoS::to_string(void) const
{
return (qos::enabled() ? "true" : "false");
}
int
QoS::from_string(const std::string_view s_)
{
int rv;
bool enable;
rv = str::from(s_,&enable);
if(rv)
return rv;
qos::enable(enable);
return 0;
}
std::string
QoSRules::to_string(void) const
{
return qos::rules_path();
}
int
QoSRules::from_string(const std::string_view s_)
{
int rv;
std::string err;
const std::string path{s_};
rv = qos::load_file(path,&err);
if(rv < 0)
{
SysLog::error("qos.rules: {}: {}",path,err);
return rv;
}
SysLog::info("qos.rules: loaded {}",path);
return 0;
}
QoSRuleSet::QoSRuleSet()
{
ro = true;
}
std::string
QoSRuleSet::to_string(void) const
{
qos::RuleSet::Ptr rs = qos::ruleset();
if(rs == nullptr)
return {};
return rs->to_string();
}
int
QoSRuleSet::from_string(const std::string_view)
{
return -EINVAL;
}
std::string
QoSStats::to_string(void) const
{
return qos::stats();
}
int
QoSStats::from_string(const std::string_view s_)
{
if(s_ != "reset")
return -EINVAL;
qos::reset_stats();
return 0;
}
std::string
QoSMaxSleepers::to_string(void) const
{
return std::to_string(qos::max_sleepers.load(std::memory_order_relaxed));
}
int
QoSMaxSleepers::from_string(const std::string_view s_)
{
int rv;
int n;
rv = str::from(s_,&n);
if(rv)
return rv;
if(n < 0)
return -EINVAL;
qos::max_sleepers.store(n,std::memory_order_relaxed);
return 0;
}
std::string
QoSMaxSleepMS::to_string(void) const
{
return std::to_string(qos::max_sleep_ns.load(std::memory_order_relaxed) /
(1000 * 1000));
}
int
QoSMaxSleepMS::from_string(const std::string_view s_)
{
int rv;
int n;
rv = str::from(s_,&n);
if(rv)
return rv;
if(n < 0)
return -EINVAL;
qos::max_sleep_ns.store(static_cast<u64>(n) * 1000 * 1000,
std::memory_order_relaxed);
return 0;
}

80
src/config_qos.hpp Normal file
View file

@ -0,0 +1,80 @@
/*
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 "tofrom_string.hpp"
// qos=<bool>
class QoS : public ToFromString
{
public:
QoS(const bool);
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.rules=<filepath>
//
// Assigning a path loads it. Assigning the same path again is how a
// ruleset is reloaded at runtime. A file that fails to parse leaves
// the running ruleset in place.
class QoSRules : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.ruleset -- read only, the ruleset as parsed.
class QoSRuleSet : public ToFromString
{
public:
QoSRuleSet();
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.stats -- counters. Assigning "reset" zeroes them.
class QoSStats : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.max-sleepers=<int>
class QoSMaxSleepers : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.max-sleep-ms=<int>
class QoSMaxSleepMS : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};

4
src/fuse_read.cpp
View file

@ -22,6 +22,7 @@
#include "fileinfo.hpp"
#include "fs_pread.hpp"
#include "ioprio.hpp"
#include "qos.hpp"
#include "state.hpp"
#include "fuse.h"
@ -66,8 +67,11 @@ FUSE::read(const fuse_req_ctx_t *ctx_,
off_t offset_)
{
ioprio::SetFrom iop(ctx_->pid);
qos::Apply q(ctx_);
FileInfo *fi;
qos::throttle(q.cls(),size_);
fi = state.get_fi(ctx_,ffi_->fh);
if(not fi)
return -EBADF;

7
src/fuse_write.cpp
View file

@ -27,6 +27,7 @@
#include "fs_pwrite.hpp"
#include "fs_pwriten.hpp"
#include "ioprio.hpp"
#include "qos.hpp"
#include "state.hpp"
#include "scope_guard/scope_guard.hpp"
@ -202,6 +203,12 @@ FUSE::write(const fuse_req_ctx_t *ctx_,
off_t offset_)
{
ioprio::SetFrom iop(ctx_->pid);
qos::Apply q(ctx_);
// Charged and, if over rate, slept for before any FileInfo lock is
// taken. Sleeping while holding fi->mutex would stall every other
// writer to the same file regardless of its class.
qos::throttle(q.cls(),count_);
return ::_write(ctx_,ffi_,buf_,count_,offset_);
}

30
src/procfs.cpp
View file

@ -105,3 +105,33 @@ procfs::get_name(const int tid_)
return commpath.data();
}
std::string
procfs::get_cgroup(const int pid_)
{
int fd;
int rv;
std::array<char,1024> buf;
fmt::format_to_n_result<char*> frv;
if(g_PROCFS_DIR_FD < 0)
fatal::abort("procfs::get_cgroup called before procfs::init()");
frv = fmt::format_to_n(buf.data(),buf.size()-1,"{}/cgroup",pid_);
frv.out[0] = '\0';
fd = fs::openat(g_PROCFS_DIR_FD,buf.data(),O_RDONLY);
if(fd < 0)
return {};
DEFER { fs::close(fd); };
rv = fs::read(fd,buf.data(),buf.size()-1);
if(rv <= 0)
return {};
if(buf[rv-1] == '\n')
rv--;
buf[rv] = '\0';
return buf.data();
}

4
src/procfs.hpp
View file

@ -27,4 +27,8 @@ namespace procfs
void init();
void shutdown();
std::string get_name(const int tid);
// Raw contents of /proc/<pid>/cgroup with the trailing newline
// removed. Empty when the process is gone or unreadable.
std::string get_cgroup(const int pid);
}

421
src/qos.cpp Normal file
View file

@ -0,0 +1,421 @@
/*
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 "qos.hpp"
#include "ioprio.hpp"
#include "procfs.hpp"
#include "fmt/core.h"
#include <errno.h>
#include <sys/resource.h>
#include <time.h>
#include <fstream>
#include <mutex>
#include <sstream>
using qos::Class;
using qos::RuleSet;
namespace qos
{
std::atomic<bool> _enabled{false};
// Deliberately small. Each sleeping process thread is one fewer
// thread draining the shared request queue, so the ceiling has to
// stay well under process-thread-count. See qos.hpp.
std::atomic<int> max_sleepers{2};
std::atomic<u64> max_sleep_ns{50ULL * 1000 * 1000};
}
namespace
{
constexpr u64 NS_PER_SEC = 1000000000ULL;
// How long a pid's classification is trusted before it is looked up
// again. Short enough that a pid recycled onto a different class is
// corrected quickly, long enough that a streaming reader is not
// re-reading /proc on every request.
constexpr u64 CACHE_TTL_NS = 10 * NS_PER_SEC;
constexpr int CACHE_SIZE = 8;
// Ceiling on how far into the future a class's refill point may be
// pushed. Without it a burst that cannot be slept off -- because
// sleeps are capped, or too many threads are already sleeping --
// would leave the class paying down debt long after the burst
// ended, throttling traffic that did nothing wrong.
constexpr u64 MAX_DEBT_NS = 2 * NS_PER_SEC;
std::mutex g_mutex;
std::atomic<u64> g_generation{1};
std::atomic<int> g_sleepers{0};
std::string g_path;
// Function-local static so the first ruleset is built on demand
// rather than during static initialisation, where its ordering
// against Config's constructor would not be defined.
RuleSet::Ptr &
_ruleset()
{
static RuleSet::Ptr rs = RuleSet::make_default();
return rs;
}
struct CacheEntry
{
u32 pid = 0;
u64 expire = 0;
const Class *cls = nullptr;
};
// Per worker thread state. Holding the RuleSet::Ptr here, rather
// than copying it per request, is what makes the bare Class
// pointers in the cache safe: the ruleset a thread is using cannot
// be freed while that thread still references it, and the thread
// only swaps to a newer one at a generation change, which also
// clears the cache.
struct TLS
{
RuleSet::Ptr rs;
u64 generation = 0;
CacheEntry cache[CACHE_SIZE];
int next = 0;
int applied_ioprio = qos::UNSET;
int applied_nice = qos::UNSET;
};
thread_local TLS tls;
u64
_now_ns()
{
struct timespec ts;
::clock_gettime(CLOCK_MONOTONIC,&ts);
return ((static_cast<u64>(ts.tv_sec) * NS_PER_SEC) +
static_cast<u64>(ts.tv_nsec));
}
void
_refresh_ruleset()
{
const u64 gen = g_generation.load(std::memory_order_acquire);
if(tls.generation == gen)
return;
{
std::lock_guard<std::mutex> lk(g_mutex);
tls.rs = ::_ruleset();
}
tls.generation = gen;
for(auto &e : tls.cache)
e = CacheEntry{};
tls.next = 0;
}
const Class *
_lookup(const u32 pid_,
const u64 now_)
{
for(const auto &e : tls.cache)
{
if((e.pid == pid_) && (now_ < e.expire))
return e.cls;
}
return nullptr;
}
void
_remember(const u32 pid_,
const u64 now_,
const Class *cls_)
{
tls.cache[tls.next] = CacheEntry{pid_,now_ + CACHE_TTL_NS,cls_};
tls.next = ((tls.next + 1) % CACHE_SIZE);
}
}
void
qos::enable(const bool enable_)
{
_enabled.store(enable_,std::memory_order_relaxed);
}
RuleSet::Ptr
qos::ruleset()
{
std::lock_guard<std::mutex> lk(g_mutex);
return ::_ruleset();
}
void
qos::set_ruleset(RuleSet::Ptr rs_)
{
{
std::lock_guard<std::mutex> lk(g_mutex);
::_ruleset() = std::move(rs_);
}
// Published after the swap so a thread that sees the new generation
// is guaranteed to see the new ruleset.
g_generation.fetch_add(1,std::memory_order_release);
}
int
qos::load_file(const std::string &path_,
std::string *err_)
{
std::ifstream ifstrm;
ifstrm.open(path_);
if(!ifstrm.good())
{
*err_ = fmt::format("unable to open {}",path_);
return -EIO;
}
std::stringstream ss;
ss << ifstrm.rdbuf();
RuleSet::Ptr rs = RuleSet::parse(ss.str(),err_);
if(rs == nullptr)
return -EINVAL;
qos::set_ruleset(std::move(rs));
{
std::lock_guard<std::mutex> lk(g_mutex);
g_path = path_;
}
return 0;
}
void
qos::throttle(const Class *cls_,
const u64 bytes_)
{
if(cls_ == nullptr)
return;
cls_->requests.fetch_add(1,std::memory_order_relaxed);
cls_->bytes.fetch_add(bytes_,std::memory_order_relaxed);
if(cls_->rate == 0)
return;
u64 sleep_ns;
{
std::lock_guard<std::mutex> lk(cls_->bucket_mutex);
const u64 now = ::_now_ns();
if(cls_->refilled == 0)
cls_->refilled = now;
if(now > cls_->refilled)
{
// 128 bit so a bucket that has been idle for a long time
// cannot overflow the refill computation. The result is
// clamped to burst regardless.
const unsigned __int128 elapsed = (now - cls_->refilled);
const unsigned __int128 added =
((elapsed * static_cast<unsigned __int128>(cls_->rate)) / NS_PER_SEC);
if(added >= cls_->burst)
cls_->tokens = cls_->burst;
else
cls_->tokens = std::min<u64>(cls_->burst,
cls_->tokens + static_cast<u64>(added));
cls_->refilled = now;
}
if(cls_->tokens >= bytes_)
{
cls_->tokens -= bytes_;
return;
}
const u64 deficit = (bytes_ - cls_->tokens);
cls_->tokens = 0;
// Carry the debt by pushing the refill point into the future
// rather than forgiving it. Tokens accrue in real time, so a
// request that sleeps off its own deficit would otherwise find
// those same tokens waiting for it -- and hand them to the next
// request as well, which lets a class run at twice its rate.
//
// Moving `refilled` past the debt spends that time in advance. It
// also means a request allowed through unthrottled below still
// owes for its bytes; the next request in the class pays.
const u64 need_ns = static_cast<u64>((static_cast<unsigned __int128>(deficit) *
NS_PER_SEC) / cls_->rate);
cls_->refilled += need_ns;
if(cls_->refilled > (now + MAX_DEBT_NS))
cls_->refilled = (now + MAX_DEBT_NS);
sleep_ns = ((cls_->refilled > now) ? (cls_->refilled - now) : 0);
}
const u64 cap = qos::max_sleep_ns.load(std::memory_order_relaxed);
if(sleep_ns > cap)
{
sleep_ns = cap;
cls_->passed.fetch_add(1,std::memory_order_relaxed);
}
if(sleep_ns == 0)
return;
const int sleepers = (g_sleepers.fetch_add(1,std::memory_order_relaxed) + 1);
if(sleepers > qos::max_sleepers.load(std::memory_order_relaxed))
{
g_sleepers.fetch_sub(1,std::memory_order_relaxed);
cls_->passed.fetch_add(1,std::memory_order_relaxed);
return;
}
struct timespec ts;
ts.tv_sec = (sleep_ns / NS_PER_SEC);
ts.tv_nsec = (sleep_ns % NS_PER_SEC);
::nanosleep(&ts,nullptr);
g_sleepers.fetch_sub(1,std::memory_order_relaxed);
cls_->throttled.fetch_add(1,std::memory_order_relaxed);
cls_->throttled_ns.fetch_add(sleep_ns,std::memory_order_relaxed);
}
void
qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_)
{
::_refresh_ruleset();
const RuleSet *rs = tls.rs.get();
if((rs == nullptr) || rs->inert())
return;
const u64 now = ::_now_ns();
const Class *cls = ::_lookup(ctx_->pid,now);
if(cls == nullptr)
{
// Only the fields some rule actually tests are read, so a
// ruleset matching on cgroup alone costs one file read per
// cache miss rather than two.
const std::string cgroup = (rs->needs_cgroup()
? procfs::get_cgroup(ctx_->pid)
: std::string{});
const std::string comm = (rs->needs_comm()
? procfs::get_name(ctx_->pid)
: std::string{});
cls = rs->classify(cgroup,comm,ctx_->uid,ctx_->gid);
::_remember(ctx_->pid,now,cls);
}
if(cls == nullptr)
return;
_cls = cls;
if((cls->ioprio != qos::UNSET) && (cls->ioprio != tls.applied_ioprio))
{
if(::ioprio::set(0,cls->ioprio) >= 0)
tls.applied_ioprio = cls->ioprio;
}
if((cls->nice != qos::UNSET) && (cls->nice != tls.applied_nice))
{
errno = 0;
if(::setpriority(PRIO_PROCESS,0,cls->nice) == 0 || errno == 0)
tls.applied_nice = cls->nice;
}
}
std::string
qos::rules_path()
{
std::lock_guard<std::mutex> lk(g_mutex);
return g_path;
}
std::string
qos::stats()
{
RuleSet::Ptr rs = qos::ruleset();
std::string out;
if(rs == nullptr)
return {};
out += fmt::format("enabled={} sleepers-in-flight={} max-sleepers={} "
"max-sleep-ms={}\n",
(qos::enabled() ? "true" : "false"),
g_sleepers.load(std::memory_order_relaxed),
qos::max_sleepers.load(std::memory_order_relaxed),
(qos::max_sleep_ns.load(std::memory_order_relaxed) /
(1000 * 1000)));
for(const auto &c : rs->classes())
{
out += fmt::format("{}: requests={} bytes={} throttled={} "
"throttled-ms={} passed={}\n",
c->name,
c->requests.load(std::memory_order_relaxed),
c->bytes.load(std::memory_order_relaxed),
c->throttled.load(std::memory_order_relaxed),
(c->throttled_ns.load(std::memory_order_relaxed) /
(1000 * 1000)),
c->passed.load(std::memory_order_relaxed));
}
return out;
}
void
qos::reset_stats()
{
RuleSet::Ptr rs = qos::ruleset();
if(rs == nullptr)
return;
for(const auto &c : rs->classes())
{
c->requests.store(0,std::memory_order_relaxed);
c->bytes.store(0,std::memory_order_relaxed);
c->throttled.store(0,std::memory_order_relaxed);
c->throttled_ns.store(0,std::memory_order_relaxed);
c->passed.store(0,std::memory_order_relaxed);
}
}

127
src/qos.hpp Normal file
View file

@ -0,0 +1,127 @@
/*
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.
*/
/*
Per-client quality of service.
mergerfs is a single daemon serving every process that touches the
pool, so from the kernel's point of view all pool I/O is issued by
one cgroup and one set of threads. Block layer priority
(ionice/BFQ, cgroup io.weight) therefore cannot tell a media player
apart from a torrent client: they are flattened into the same
issuer.
This restores the distinction inside the daemon. Each request is
attributed to the calling process, matched against a ruleset, and
the worker thread is given the ioprio and nice value of the class it
matched for the duration of the request. Optionally the class is
also rate limited.
Compared to the `proxy_ioprio` option, which copies whatever ioprio
the caller happens to have, the policy lives in one place and does
not depend on every client being correctly ioniced by whoever
started it -- which matters most for clients that respawn workers,
or that live in a container the daemon does not control.
*/
#pragma once
#include "qos_class.hpp"
#include "qos_rules.hpp"
#include "fuse_req_ctx.h"
#include <atomic>
#include <string>
namespace qos
{
// Exposed so the disabled-path check inlines into every
// fuse_read/fuse_write as an atomic load and a branch.
extern std::atomic<bool> _enabled;
[[gnu::always_inline]]
inline
bool
enabled()
{
return _enabled.load(std::memory_order_relaxed);
}
void enable(const bool);
// Replaces the active ruleset. Safe to call while requests are in
// flight.
void set_ruleset(RuleSet::Ptr);
RuleSet::Ptr ruleset();
// Loads and installs a ruleset from `path`. On failure the existing
// ruleset is left untouched and `err` describes the problem.
int load_file(const std::string &path, std::string *err);
// Number of process threads allowed to be sleeping in a throttle at
// once, and the longest any single request may be delayed.
//
// Both exist to bound the damage throttling can do. mergerfs hands
// requests from its read threads to a *bounded* process thread
// queue, so a sleeping process thread is a process thread not
// serving anyone -- and once the queue backs up, requests of every
// class queue behind it. Rather than let a rate limit turn into a
// pool-wide stall, a request that would exceed either bound is
// allowed through unthrottled and counted in the `passed` stat.
extern std::atomic<int> max_sleepers;
extern std::atomic<u64> max_sleep_ns;
// Path most recently loaded by load_file(), or empty.
std::string rules_path();
std::string stats();
void reset_stats();
// Charges `bytes` to the class and, if that class is over its rate,
// sleeps for as long as the bounds above permit. Called with the
// size the caller asked for, before the I/O is issued.
void throttle(const Class *, const u64 bytes);
// Classifies the calling process and applies its class to this
// thread. The thread keeps those settings until another request
// changes them, so nothing is restored on destruction -- FUSE
// worker threads do nothing between requests worth protecting, and
// restoring would double the syscalls on the hot path.
class Apply
{
public:
[[gnu::always_inline]]
inline
Apply(const fuse_req_ctx_t *ctx_)
{
if(qos::enabled())
_slow_apply(ctx_);
}
// nullptr when QoS is off or the ruleset cannot change anything.
const Class *cls() const { return _cls; }
private:
void _slow_apply(const fuse_req_ctx_t *);
private:
const Class *_cls = nullptr;
};
}

119
src/qos_class.hpp Normal file
View file

@ -0,0 +1,119 @@
/*
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"
#include <atomic>
#include <mutex>
#include <string>
namespace qos
{
// Sentinel meaning "leave the worker thread's current value alone".
constexpr int UNSET = INT32_MIN;
namespace ioprio
{
// Mirrors the kernel's linux/ioprio.h encoding. Redefined here so
// mergerfs does not depend on a UAPI header that has moved
// between kernel versions.
constexpr int CLASS_SHIFT = 13;
constexpr int PRIO_MASK = (1 << CLASS_SHIFT) - 1;
constexpr int CLASS_NONE = 0;
constexpr int CLASS_RT = 1;
constexpr int CLASS_BE = 2;
constexpr int CLASS_IDLE = 3;
constexpr
int
value(const int class_,
const int data_)
{
return ((class_ << CLASS_SHIFT) | data_);
}
constexpr
int
to_class(const int value_)
{
return (value_ >> CLASS_SHIFT);
}
constexpr
int
to_data(const int value_)
{
return (value_ & PRIO_MASK);
}
std::string to_string(const int value);
// Parses "rt:N", "be:N", "idle" and "none". Returns 0 on success
// and -EINVAL otherwise.
int from_string(const std::string_view, int *value);
}
// A QoS class is a named bundle of scheduling parameters plus the
// token bucket and counters belonging to every request matched to
// it.
//
// The mutable atomics are the only members written after a ruleset
// is published; the descriptive members are immutable for the life
// of the object, which is what lets the classifier hand out bare
// pointers to it.
class Class
{
public:
Class(std::string name_)
: name(std::move(name_))
{
}
public:
std::string name;
int ioprio = qos::UNSET;
int nice = qos::UNSET;
// 0 == unlimited. `burst` is the bucket depth and defaults to one
// second of `rate` when a rate is set.
u64 rate = 0;
u64 burst = 0;
public:
// Token bucket, guarded by `bucket_mutex`. `tokens` is in bytes
// and `refilled` is a CLOCK_MONOTONIC nanosecond stamp. All three
// are only touched when `rate` is non-zero, so classes without a
// rate never take the lock.
mutable std::mutex bucket_mutex;
mutable u64 tokens{0};
mutable u64 refilled{0};
public:
// Counters, reported via the qos.stats config key.
mutable std::atomic<u64> requests{0};
mutable std::atomic<u64> bytes{0};
mutable std::atomic<u64> throttled{0};
mutable std::atomic<u64> throttled_ns{0};
mutable std::atomic<u64> passed{0};
};
}

454
src/qos_rules.cpp Normal file
View file

@ -0,0 +1,454 @@
/*
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 "qos_rules.hpp"
#include "fmt/core.h"
#include <errno.h>
#include <fnmatch.h>
#include <cctype>
#include <cstdlib>
#include <sstream>
using qos::Class;
using qos::Field;
using qos::Op;
using qos::Rule;
using qos::RuleSet;
std::string
qos::ioprio::to_string(const int value_)
{
if(value_ == qos::UNSET)
return "unset";
switch(qos::ioprio::to_class(value_))
{
case qos::ioprio::CLASS_NONE:
return "none";
case qos::ioprio::CLASS_RT:
return fmt::format("rt:{}",qos::ioprio::to_data(value_));
case qos::ioprio::CLASS_BE:
return fmt::format("be:{}",qos::ioprio::to_data(value_));
case qos::ioprio::CLASS_IDLE:
return "idle";
}
return "unset";
}
int
qos::ioprio::from_string(const std::string_view s_,
int *value_)
{
if(s_ == "idle")
{
// The idle class carries no priority data.
*value_ = qos::ioprio::value(qos::ioprio::CLASS_IDLE,0);
return 0;
}
if(s_ == "none")
{
*value_ = qos::ioprio::value(qos::ioprio::CLASS_NONE,0);
return 0;
}
int klass;
if(s_.rfind("rt:",0) == 0)
klass = qos::ioprio::CLASS_RT;
else if(s_.rfind("be:",0) == 0)
klass = qos::ioprio::CLASS_BE;
else
return -EINVAL;
const std::string_view data = s_.substr(3);
if((data.size() != 1) || !std::isdigit(static_cast<unsigned char>(data[0])))
return -EINVAL;
const int n = (data[0] - '0');
if(n > 7)
return -EINVAL;
*value_ = qos::ioprio::value(klass,n);
return 0;
}
int
qos::parse_size(const std::string_view s_,
u64 *bytes_)
{
if(s_.empty())
return -EINVAL;
std::size_t i = 0;
while((i < s_.size()) &&
(std::isdigit(static_cast<unsigned char>(s_[i])) || (s_[i] == '.')))
i++;
if(i == 0)
return -EINVAL;
const std::string num(s_.substr(0,i));
errno = 0;
char *end = nullptr;
const double value = ::strtod(num.c_str(),&end);
if(errno || (end == num.c_str()) || (*end != '\0') || (value < 0))
return -EINVAL;
std::string suffix(s_.substr(i));
// Accept "MB" and "MiB" as aliases of "M". mergerfs treats all size
// suffixes as binary multiples, matching minfreespace and friends.
if((suffix.size() > 1) && (suffix.back() == 'B'))
suffix.pop_back();
if((suffix.size() > 1) && (suffix.back() == 'i'))
suffix.pop_back();
u64 mult;
if(suffix.empty())
mult = 1ULL;
else if((suffix == "K") || (suffix == "k"))
mult = 1024ULL;
else if((suffix == "M") || (suffix == "m"))
mult = 1024ULL * 1024;
else if((suffix == "G") || (suffix == "g"))
mult = 1024ULL * 1024 * 1024;
else if((suffix == "T") || (suffix == "t"))
mult = 1024ULL * 1024 * 1024 * 1024;
else
return -EINVAL;
*bytes_ = static_cast<u64>(value * static_cast<double>(mult));
return 0;
}
bool
Rule::matches(const std::string &cgroup_,
const std::string &comm_,
const u32 uid_,
const u32 gid_) const
{
switch(field)
{
case Field::UID:
return (uid_ == number);
case Field::GID:
return (gid_ == number);
case Field::CGROUP:
case Field::COMM:
break;
}
const std::string &subject = ((field == Field::CGROUP) ? cgroup_ : comm_);
if(op == Op::EQ)
return (subject == pattern);
return (::fnmatch(pattern.c_str(),subject.c_str(),0) == 0);
}
const Class *
RuleSet::find_class(const std::string_view name_) const
{
for(const auto &c : _classes)
{
if(c->name == name_)
return c.get();
}
return nullptr;
}
const Class *
RuleSet::classify(const std::string &cgroup_,
const std::string &comm_,
const u32 uid_,
const u32 gid_) const
{
for(const auto &rule : _rules)
{
if(rule.matches(cgroup_,comm_,uid_,gid_))
return rule.cls;
}
return _default;
}
RuleSet::Ptr
RuleSet::make_default()
{
auto rs = std::make_shared<RuleSet>();
rs->_classes.push_back(std::make_unique<Class>("default"));
rs->_default = rs->_classes.back().get();
rs->_inert = true;
return rs;
}
namespace
{
std::vector<std::string>
tokenize(const std::string &line_)
{
std::vector<std::string> tokens;
std::istringstream iss(line_);
std::string tok;
while(iss >> tok)
tokens.push_back(tok);
return tokens;
}
int
parse_field(const std::string &s_,
Field *field_)
{
if(s_ == "cgroup")
*field_ = Field::CGROUP;
else if(s_ == "comm")
*field_ = Field::COMM;
else if(s_ == "uid")
*field_ = Field::UID;
else if(s_ == "gid")
*field_ = Field::GID;
else
return -EINVAL;
return 0;
}
}
RuleSet::Ptr
RuleSet::parse(const std::string_view text_,
std::string *err_)
{
auto rs = std::make_shared<RuleSet>();
std::istringstream istrm{std::string(text_)};
std::string line;
std::string default_name;
int lineno = 0;
auto fail =
[&](const std::string &msg_) -> RuleSet::Ptr
{
*err_ = fmt::format("line {}: {}",lineno,msg_);
return nullptr;
};
while(std::getline(istrm,line,'\n'))
{
lineno++;
const std::size_t hash = line.find('#');
if(hash != std::string::npos)
line.erase(hash);
auto tokens = ::tokenize(line);
if(tokens.empty())
continue;
if(tokens[0] == "class")
{
if(tokens.size() < 2)
return fail("'class' needs a name");
if(rs->find_class(tokens[1]))
return fail(fmt::format("duplicate class '{}'",tokens[1]));
auto cls = std::make_unique<Class>(tokens[1]);
for(std::size_t i = 2; i < tokens.size(); i++)
{
const std::size_t eq = tokens[i].find('=');
if(eq == std::string::npos)
return fail(fmt::format("expected key=value, got '{}'",tokens[i]));
const std::string key = tokens[i].substr(0,eq);
const std::string val = tokens[i].substr(eq + 1);
if(key == "ioprio")
{
if(qos::ioprio::from_string(val,&cls->ioprio))
return fail(fmt::format("invalid ioprio '{}'",val));
}
else if(key == "nice")
{
const int n = std::atoi(val.c_str());
if((n < -20) || (n > 19))
return fail(fmt::format("nice out of range: '{}'",val));
cls->nice = n;
}
else if(key == "rate")
{
if(qos::parse_size(val,&cls->rate))
return fail(fmt::format("invalid rate '{}'",val));
}
else if(key == "burst")
{
if(qos::parse_size(val,&cls->burst))
return fail(fmt::format("invalid burst '{}'",val));
}
else
{
return fail(fmt::format("unknown class key '{}'",key));
}
}
// A bucket with no explicit depth holds one second of
// traffic, which is enough to absorb a readahead burst
// without letting an idle class bank unlimited credit.
if(cls->rate && (cls->burst == 0))
cls->burst = cls->rate;
if(cls->burst < cls->rate)
cls->burst = cls->rate;
cls->tokens = cls->burst;
if((cls->ioprio != qos::UNSET) ||
(cls->nice != qos::UNSET) ||
(cls->rate != 0))
rs->_inert = false;
rs->_classes.push_back(std::move(cls));
}
else if(tokens[0] == "match")
{
if(tokens.size() != 6)
return fail("expected: match <field> <=|~> <pattern> -> <class>");
if(tokens[4] != "->")
return fail(fmt::format("expected '->', got '{}'",tokens[4]));
Rule rule{};
if(::parse_field(tokens[1],&rule.field))
return fail(fmt::format("unknown field '{}'",tokens[1]));
if(tokens[2] == "=")
rule.op = Op::EQ;
else if(tokens[2] == "~")
rule.op = Op::GLOB;
else
return fail(fmt::format("unknown operator '{}'",tokens[2]));
if((rule.field == Field::UID) || (rule.field == Field::GID))
{
if(rule.op != Op::EQ)
return fail("uid/gid only support the '=' operator");
char *end = nullptr;
const unsigned long n = ::strtoul(tokens[3].c_str(),&end,10);
if((end == tokens[3].c_str()) || (*end != '\0'))
return fail(fmt::format("invalid uid/gid '{}'",tokens[3]));
rule.number = static_cast<u32>(n);
}
rule.pattern = tokens[3];
rule.cls = rs->find_class(tokens[5]);
if(rule.cls == nullptr)
return fail(fmt::format("unknown class '{}' (classes must be "
"defined before use)",tokens[5]));
if(rule.field == Field::CGROUP)
rs->_needs_cgroup = true;
if(rule.field == Field::COMM)
rs->_needs_comm = true;
rs->_rules.push_back(std::move(rule));
}
else if(tokens[0] == "default")
{
if(tokens.size() != 2)
return fail("expected: default <class>");
default_name = tokens[1];
}
else
{
return fail(fmt::format("unknown directive '{}'",tokens[0]));
}
}
if(default_name.empty())
{
// No explicit default: unmatched processes keep whatever
// priority they already had.
rs->_classes.push_back(std::make_unique<Class>("unclassified"));
rs->_default = rs->_classes.back().get();
}
else
{
rs->_default = rs->find_class(default_name);
if(rs->_default == nullptr)
{
*err_ = fmt::format("default names unknown class '{}'",default_name);
return nullptr;
}
}
if(rs->_rules.empty() && rs->_default->ioprio == qos::UNSET &&
rs->_default->nice == qos::UNSET && rs->_default->rate == 0)
rs->_inert = true;
return rs;
}
std::string
RuleSet::to_string() const
{
std::string s;
for(const auto &c : _classes)
{
s += fmt::format("class {} ioprio={} nice={} rate={} burst={}\n",
c->name,
qos::ioprio::to_string(c->ioprio),
((c->nice == qos::UNSET) ? "unset" : std::to_string(c->nice)),
c->rate,
c->burst);
}
for(const auto &r : _rules)
{
const char *field = "";
switch(r.field)
{
case Field::CGROUP: field = "cgroup"; break;
case Field::COMM: field = "comm"; break;
case Field::UID: field = "uid"; break;
case Field::GID: field = "gid"; break;
}
s += fmt::format("match {} {} {} -> {}\n",
field,
((r.op == Op::EQ) ? "=" : "~"),
r.pattern,
r.cls->name);
}
if(_default)
s += fmt::format("default {}\n",_default->name);
return s;
}

113
src/qos_rules.hpp Normal file
View file

@ -0,0 +1,113 @@
/*
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 "qos_class.hpp"
#include <memory>
#include <string>
#include <vector>
namespace qos
{
// Which attribute of the calling process a rule tests.
enum class Field
{
CGROUP, // contents of /proc/<pid>/cgroup
COMM, // contents of /proc/<pid>/comm
UID,
GID
};
enum class Op
{
EQ, // '=' exact string / numeric equality
GLOB // '~' fnmatch(3) pattern
};
struct Rule
{
Field field;
Op op;
std::string pattern;
u32 number; // parsed form of `pattern` for UID/GID
const Class *cls; // owned by the RuleSet holding this rule
bool matches(const std::string &cgroup,
const std::string &comm,
const u32 uid,
const u32 gid) const;
};
// An immutable, atomically published set of classes and rules.
//
// Loading a rules file builds a whole new RuleSet and swaps it in,
// so a reload can never expose a half-applied configuration to a
// request in flight. Readers keep a shared_ptr for the duration of
// a request, which is what keeps a Class alive after a swap.
class RuleSet
{
public:
typedef std::shared_ptr<const RuleSet> Ptr;
public:
// Builds the ruleset every mount starts with: a single
// pass-through class and no rules, which leaves behaviour
// identical to qos being off.
static Ptr make_default();
// Parses `text` and returns the new ruleset, or nullptr on
// error, in which case `err` describes the first problem found
// and the caller must keep using the ruleset it already has.
static Ptr parse(const std::string_view text,
std::string *err);
public:
const Class *classify(const std::string &cgroup,
const std::string &comm,
const u32 uid,
const u32 gid) const;
const Class *find_class(const std::string_view name) const;
// True when no rule and no class can change a thread's behaviour,
// letting the hot path skip reading /proc entirely.
bool inert() const { return _inert; }
bool needs_cgroup() const { return _needs_cgroup; }
bool needs_comm() const { return _needs_comm; }
const std::vector<std::unique_ptr<Class>> &classes() const { return _classes; }
std::string to_string() const;
private:
std::vector<std::unique_ptr<Class>> _classes;
std::vector<Rule> _rules;
const Class *_default = nullptr;
bool _inert = true;
bool _needs_cgroup = false;
bool _needs_comm = false;
};
// Parses "10M", "1.5MiB", "512K", "1G" and bare byte counts into
// bytes. Returns 0 on success, -EINVAL otherwise.
int parse_size(const std::string_view, u64 *bytes);
}