diff --git a/src/config.cpp b/src/config.cpp index 2194c1dc..68539f77 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -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; diff --git a/src/config.hpp b/src/config.hpp index 0005a2e3..cf1a20a6 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -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 process_thread_count; TFSRef 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 read_thread_count; ConfigU64 readahead; FUSE::ReadDir readdir; diff --git a/src/config_qos.cpp b/src/config_qos.cpp new file mode 100644 index 00000000..72d5091e --- /dev/null +++ b/src/config_qos.cpp @@ -0,0 +1,164 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 + + +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(n) * 1000 * 1000, + std::memory_order_relaxed); + + return 0; +} diff --git a/src/config_qos.hpp b/src/config_qos.hpp new file mode 100644 index 00000000..0a58a51d --- /dev/null +++ b/src/config_qos.hpp @@ -0,0 +1,80 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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= +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= +// +// 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= +class QoSMaxSleepers : public ToFromString +{ +public: + std::string to_string(void) const final; + int from_string(const std::string_view) final; +}; + +// qos.max-sleep-ms= +class QoSMaxSleepMS : public ToFromString +{ +public: + std::string to_string(void) const final; + int from_string(const std::string_view) final; +}; diff --git a/src/fuse_read.cpp b/src/fuse_read.cpp index 57c9daae..e3d5209a 100644 --- a/src/fuse_read.cpp +++ b/src/fuse_read.cpp @@ -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; diff --git a/src/fuse_write.cpp b/src/fuse_write.cpp index b91c7ba0..e6916d85 100644 --- a/src/fuse_write.cpp +++ b/src/fuse_write.cpp @@ -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_); } diff --git a/src/procfs.cpp b/src/procfs.cpp index e00bb296..7d2bca9b 100644 --- a/src/procfs.cpp +++ b/src/procfs.cpp @@ -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 buf; + fmt::format_to_n_result 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(); +} diff --git a/src/procfs.hpp b/src/procfs.hpp index 22167718..b3907f61 100644 --- a/src/procfs.hpp +++ b/src/procfs.hpp @@ -27,4 +27,8 @@ namespace procfs void init(); void shutdown(); std::string get_name(const int tid); + + // Raw contents of /proc//cgroup with the trailing newline + // removed. Empty when the process is gone or unreadable. + std::string get_cgroup(const int pid); } diff --git a/src/qos.cpp b/src/qos.cpp new file mode 100644 index 00000000..97e6f707 --- /dev/null +++ b/src/qos.cpp @@ -0,0 +1,421 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 +#include +#include + +#include +#include +#include + +using qos::Class; +using qos::RuleSet; + +namespace qos +{ + std::atomic _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 max_sleepers{2}; + std::atomic 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 g_generation{1}; + std::atomic 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(ts.tv_sec) * NS_PER_SEC) + + static_cast(ts.tv_nsec)); + } + + void + _refresh_ruleset() + { + const u64 gen = g_generation.load(std::memory_order_acquire); + + if(tls.generation == gen) + return; + + { + std::lock_guard 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 lk(g_mutex); + + return ::_ruleset(); +} + +void +qos::set_ruleset(RuleSet::Ptr rs_) +{ + { + std::lock_guard 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 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 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(cls_->rate)) / NS_PER_SEC); + + if(added >= cls_->burst) + cls_->tokens = cls_->burst; + else + cls_->tokens = std::min(cls_->burst, + cls_->tokens + static_cast(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((static_cast(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 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); + } +} diff --git a/src/qos.hpp b/src/qos.hpp new file mode 100644 index 00000000..9ba47a91 --- /dev/null +++ b/src/qos.hpp @@ -0,0 +1,127 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 +#include + + +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 _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 max_sleepers; + extern std::atomic 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; + }; +} diff --git a/src/qos_class.hpp b/src/qos_class.hpp new file mode 100644 index 00000000..03f82622 --- /dev/null +++ b/src/qos_class.hpp @@ -0,0 +1,119 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 +#include +#include + + +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 requests{0}; + mutable std::atomic bytes{0}; + mutable std::atomic throttled{0}; + mutable std::atomic throttled_ns{0}; + mutable std::atomic passed{0}; + }; +} diff --git a/src/qos_rules.cpp b/src/qos_rules.cpp new file mode 100644 index 00000000..bbd31fac --- /dev/null +++ b/src/qos_rules.cpp @@ -0,0 +1,454 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 +#include + +#include +#include +#include + +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(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(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(value * static_cast(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(); + + rs->_classes.push_back(std::make_unique("default")); + rs->_default = rs->_classes.back().get(); + rs->_inert = true; + + return rs; +} + +namespace +{ + std::vector + tokenize(const std::string &line_) + { + std::vector 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(); + 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(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 <=|~> -> "); + 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(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 "); + 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("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; +} diff --git a/src/qos_rules.hpp b/src/qos_rules.hpp new file mode 100644 index 00000000..5afec82d --- /dev/null +++ b/src/qos_rules.hpp @@ -0,0 +1,113 @@ +/* + ISC License + + Copyright (c) 2026, Antonio SJ Musumeci + + 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 +#include +#include + + +namespace qos +{ + // Which attribute of the calling process a rule tests. + enum class Field + { + CGROUP, // contents of /proc//cgroup + COMM, // contents of /proc//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 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> &classes() const { return _classes; } + + std::string to_string() const; + + private: + std::vector> _classes; + std::vector _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); +}