From 34e57dbb8a6365a14de011a17832a02280ac8fa1 Mon Sep 17 00:00:00 2001 From: Grioghar Date: Wed, 9 Sep 2026 16:34:55 -0500 Subject: [PATCH 1/2] qos: process-classified I/O priority and rate limiting 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 --- src/config.cpp | 7 + src/config.hpp | 7 + src/config_qos.cpp | 164 ++++++++++++++++ src/config_qos.hpp | 80 ++++++++ src/fuse_read.cpp | 4 + src/fuse_write.cpp | 7 + src/procfs.cpp | 30 +++ src/procfs.hpp | 4 + src/qos.cpp | 421 +++++++++++++++++++++++++++++++++++++++++ src/qos.hpp | 127 +++++++++++++ src/qos_class.hpp | 119 ++++++++++++ src/qos_rules.cpp | 454 +++++++++++++++++++++++++++++++++++++++++++++ src/qos_rules.hpp | 113 +++++++++++ 13 files changed, 1537 insertions(+) create mode 100644 src/config_qos.cpp create mode 100644 src/config_qos.hpp create mode 100644 src/qos.cpp create mode 100644 src/qos.hpp create mode 100644 src/qos_class.hpp create mode 100644 src/qos_rules.cpp create mode 100644 src/qos_rules.hpp 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); +} From 5f6275baeaeb5c1bcd545ff4467e4a5fe001221b Mon Sep 17 00:00:00 2001 From: Grioghar Date: Wed, 9 Sep 2026 17:21:57 -0500 Subject: [PATCH 2/2] qos: match on cmdline, and never throttle critical dependencies Two gaps found while using this against a real media server. Some programs do very different jobs under one name. Plex runs credits detection with the same 'Plex Transcoder' binary that serves playback, reading the same file, so neither comm nor path separates them. Adds a cmdline match field, and quoting so a pattern can contain spaces -- without which 'Plex Transcoder' cannot be written down at all. Adds a 'critical' class flag for services playback synchronously waits on. Plex's EasyAudioEncoder is the case: a transcode of EAC3 audio writes to EAE and blocks. Such a helper looks like background work, so a governor protecting playback will throttle or suspend it and hang the stream waiting behind it. A critical class is never delayed. Also fixes qos.max-sleepers auto-sizing, which read the requested thread count rather than the resolved one and so capped at 1. Co-Authored-By: Claude Opus 5 --- mkdocs/docs/config/qos.md | 329 ++++++++++++++++++++++ mkdocs/mkdocs.yml | 1 + src/config.cpp | 2 + src/config.hpp | 2 + src/config_qos.cpp | 54 +++- src/config_qos.hpp | 16 ++ src/fuse_read.cpp | 24 +- src/fuse_write.cpp | 33 ++- src/procfs.cpp | 39 +++ src/procfs.hpp | 6 + src/qos.cpp | 450 ++++++++++++++++++++++++++---- src/qos.hpp | 68 ++++- src/qos_class.cpp | 37 +++ src/qos_class.hpp | 116 +++++++- src/qos_rules.cpp | 414 ++++++++++++++++++++++----- src/qos_rules.hpp | 111 ++++++-- tests/tests.cpp | 462 +++++++++++++++++++++++++++++++ tools/mergerfs.qos-bench | 314 +++++++++++++++++++++ tools/mergerfs.qos-playback-test | 324 ++++++++++++++++++++++ 19 files changed, 2608 insertions(+), 194 deletions(-) create mode 100644 mkdocs/docs/config/qos.md create mode 100644 src/qos_class.cpp create mode 100755 tools/mergerfs.qos-bench create mode 100755 tools/mergerfs.qos-playback-test diff --git a/mkdocs/docs/config/qos.md b/mkdocs/docs/config/qos.md new file mode 100644 index 00000000..f30299c2 --- /dev/null +++ b/mkdocs/docs/config/qos.md @@ -0,0 +1,329 @@ +# qos + +* type: `BOOL` +* default: `false` +* example: `qos=true` + +Per-client quality of service: attribute each read and write to the +process that asked for it, match it against a ruleset, and give the +worker thread that class's I/O priority, nice value and bandwidth +allowance for the duration of the request. + + +## Why this exists + +mergerfs is one daemon serving every process that touches the pool. To +the kernel, all pool I/O is issued by one set of threads in one cgroup, +so block layer priority cannot tell a media player apart from a torrent +client -- `ionice`, systemd's `IOSchedulingClass=` and cgroup +`io.weight` are all applied to the *daemon*, not to whoever is behind +the request. Priority is flattened. + +This restores the distinction inside the daemon, where the caller is +still known. + +[proxy-ioprio](proxy-ioprio.md) addresses the same problem from the +other end: it copies whatever ioprio the caller happens to have. That +is simpler, but it only works if every client is correctly `ionice`d by +whoever started it -- which is awkward for programs that respawn +workers under a single service, and impossible for clients inside a +container the daemon does not control. With `qos` the policy lives in +one file and is written in terms of *who is asking* and *what they are +touching*. + +The two are independent and there is no reason to enable both. + + +## Options + +| option | default | meaning | +|---|---|---| +| `qos` | `false` | master switch | +| `qos.rules` | none | path to a rules file; assigning it loads it | +| `qos.ruleset` | - | read only: the ruleset as parsed | +| `qos.stats` | - | counters; assign `reset` to zero them | +| `qos.max-sleepers` | `-1` | threads that may sleep in a throttle at once; `-1` is half the process thread pool | +| `qos.max-sleep-ms` | `50` | longest any one request may be delayed | +| `qos.distress-ms` | `50` | service time below which a protected class is never considered to be suffering | +| `qos.distress-factor` | `3.0` | multiple of a resource's quiet latency that counts as distress | + +Every one of these is readable and writable at runtime through the +[runtime interface](../runtime_interface.md): + +```sh +getfattr -n user.mergerfs.qos.stats --only-values /mnt/pool/.mergerfs +setfattr -n user.mergerfs.qos.rules -v /etc/mergerfs/qos.rules /mnt/pool/.mergerfs +``` + +Assigning `qos.rules` is also how a ruleset is reloaded. A file that +fails to parse is rejected with the offending line number logged to +syslog, and the running ruleset is left untouched. + + +## The rules file + +A single mount option pointing at a file, rather than a value crammed +into `/etc/fstab`, because rules contain commas and fstab does not +forgive that. + +``` +# Everything after # is a comment. + +capacity /mnt/disk1 180M # measured throughput of a resource +capacity default 120M # fallback for unlisted resources + +class [protect] [ioprio=…] [nice=…] [rate=…] [burst=…] [yield=…] [floor=…] + +match [ …] -> + +default +``` + +Rules are evaluated in order and the first whose conditions *all* match +wins. A class must be defined before a rule names it. + +### Class attributes + +| attribute | meaning | +|---|---| +| `ioprio=` | `rt:0`-`rt:7`, `be:0`-`be:7`, `idle`, `none` | +| `nice=` | `-20` to `19` | +| `rate=` | absolute (`8M`) or a share of the resource's capacity (`10%`) | +| `burst=` | bucket depth; defaults to one second of `rate` | +| `protect` | never throttled, and its latency drives the governor | +| `critical` | never throttled, and *not* a control signal | +| `yield=` | `0`-`100`: how hard this class gives way under pressure | +| `floor=` | never back off below this; absolute or a percentage | + +### Match fields + +| field | matched against | +|---|---| +| `cgroup` | `/proc//cgroup` -- identifies the container | +| `comm` | `/proc//comm` -- identifies the program | +| `cmdline` | `/proc//cmdline`, arguments joined by spaces | +| `uid`, `gid` | the caller's credentials | +| `path` | the path within the pool | +| `op` | `read` or `write` | + +Operators are `=` for exact match and `~` for an +[fnmatch(3)](https://man7.org/linux/man-pages/man3/fnmatch.3.html) +glob. `FNM_PATHNAME` is not set, so `path ~ /TV/*` covers everything +beneath `/TV`. + +`uid`, `gid` and `op` accept only `=`. + +Wrap a pattern in single or double quotes to keep spaces in it. This is +not decoration -- the names worth matching include `"Plex Transcoder"` +and `"Plex Media Scanner"`, and a command line is nothing but spaces. + + +### Why `cmdline` exists + +Some programs do very different jobs under one name. Plex runs credits +and intro detection using the *same* `Plex Transcoder` binary that +serves playback, reading the *same* media file. Neither `comm` nor +`path` can tell those apart. The command line can: a detection job +writes into `.../Transcode/Detection/`, a playback transcode feeds a +session. + +``` +match comm = "Plex Transcoder" cmdline ~ */Transcode/Detection/* -> analysis +match comm = "Plex Transcoder" -> playback +``` + +Measured on an otherwise idle pool, with `analysis` capped at 8 MiB/s: +the detection job ran at 8.1 MiB/s and the playback job, same binary and +same file, at 730 MiB/s. + + +### Why `critical` exists + +`critical` marks a class that playback *synchronously waits on*, as +opposed to playback itself. Plex's EasyAudioEncoder is the canonical +case: transcoding EAC3, TrueHD or DTS audio runs +`Plex Transcoder -codec:1 eac3_eae`, which hands the audio to EAE and +blocks until it answers. + +Such a helper looks like background work -- it is not the stream, it +does not appear in a session -- but slowing it slows the stream stalled +behind it. That is a priority inversion, and it is an easy one to +create by accident: a governor that suspends "background" processes to +protect playback will happily suspend the one service playback is +waiting for, and hang the very thing it was protecting. + +A `critical` class is never delayed, whatever else the ruleset says +about it, and its latency is not used as a control signal. + +``` +class eae critical +match comm ~ EasyAudioEncode* -> eae +match comm = "Plex EAE Service" -> eae +``` + + +## Rates are per resource + +A resource is a branch -- in practice, a disk. Each class gets its own +token bucket per branch, so a class limited to 10% gets a tenth of +*each* disk rather than a tenth of the pool split across all of them. +Two players reading two different disks never compete for one +allowance. + +Percentages need a capacity to resolve against. Measure it: + +```sh +mergerfs.qos-bench /mnt/pool -o /etc/mergerfs/qos.capacity +``` + +and include those lines in the rules file. A percentage with no +capacity declared for its resource is left unlimited rather than +throttled against a guess. + + +## The adaptive governor + +A fixed cap is a blunt instrument: it slows downloads even when nothing +is playing. Marking a class `protect` turns on a feedback loop instead. + +mergerfs times that class's requests per resource and compares the +smoothed figure against the quietest that resource has managed. When it +is worse by `qos.distress-factor` -- and worse than `qos.distress-ms` +in absolute terms -- *and* a yielding class is active on that same +resource, pressure rises. Every yielding class's allowance is then +scaled by `1 - (pressure × yield / 100)`, bounded below by its `floor`. + +When the latency recovers, or playback stops, pressure decays back to +zero and the allowances return to whatever they were configured for -- +which for downloads is normally no limit at all. + +The result: bulk traffic runs flat out on a disk nobody is streaming +from, gives way gradually on one that is, and never stops entirely. + +Pressure rises multiplicatively and falls additively. A stutter has +already been heard by the time it is measured, so the loop backs off +hard and returns gently. + +`qos.distress-ms` is the knob that matters and it is device dependent. +The `50` default suits spinning disks, where 50ms is a genuine stall. +On an SSD or NVMe pool it will never be reached and the governor will +never engage -- use single-digit milliseconds there. Setting it to `0` +removes noise suppression entirely and is not recommended. + + +## Example + +``` +capacity default 150M + +# Playback is what we protect. Never throttled; its latency is the +# signal everything else is governed by. +class playback protect ioprio=rt:0 nice=-5 + +# Services playback blocks on. Never throttled, never a signal. +class helpers critical + +# A media server's own background work -- scans, thumbnails, chapter +# and credits detection -- yields, but gently. +class scanning yield=60 floor=10% ioprio=be:6 nice=10 + +# Downloads yield first and hardest, but never stop. +class downloads yield=100 floor=5% ioprio=idle nice=19 + +# Audio helpers first: a transcode blocks on these. +match comm ~ EasyAudioEncode* -> helpers +match comm = "Plex EAE Service" -> helpers + +# A media server's analysis work runs under the same binary as +# playback; only the command line separates them. +match comm = "Plex Transcoder" cmdline ~ */Transcode/Detection/* -> scanning +match comm = "Plex Media Scanner" -> scanning +match comm ~ ffdetect -> scanning + +# Real transcodes and direct play, across every server. +match comm = "Plex Transcoder" -> playback +match comm ~ *ffmpeg* -> playback +match comm ~ jellyfin -> playback +match comm ~ EmbyServer -> playback + +# Anything else those containers do. +match cgroup ~ *lxc/311[123]* -> scanning + +# Downloaders. +match comm ~ qbittorrent* -> downloads +match comm ~ sabnzbd* -> downloads + +default scanning +``` + +A client that reads the pool over NFS or SMB -- Kodi on another +machine, typically -- arrives as `nfsd` or `smbd`, not as itself. +Match those, or match on `path`, since the process behind the export is +not the one you care about. + + +## Testing it + +A ruleset that looks right is not the same as playback that does not +stutter. `mergerfs.qos-playback-test` models a player: it reads a real +file out of the pool at a fixed bitrate through a jitter buffer, starts +and stops competing load partway through, and counts the moments the +buffer ran dry. + +```sh +mergerfs.qos-playback-test /mnt/pool/Movies/film.mkv \ + --bulk-file /mnt/pool/TV/something.mkv \ + -b 25M --buffer 5 -d 60 --bulk-start 10 --bulk-stop 45 \ + -m /mnt/pool +``` + +It exits non-zero if the buffer ever emptied. The `pressure` column +shows the governor engaging and releasing. + + +## Limitations + +**Buffered writes are attributed to the kernel, not the writer.** A +write that lands in the page cache is flushed later by kernel writeback +threads, which are not the calling process and carry none of its +identity. Those requests fall to the `default` class. Rate limiting a +writer works because the limit is applied when the write *enters* +mergerfs; ioprio on the worker thread does not, for the same reason it +does not for any other buffered write. + +**Throttling is bounded, and therefore best effort.** mergerfs hands +requests from its read threads to a bounded process thread queue. A +sleeping process thread is one not serving anyone, and once that queue +backs up, every class queues behind it. So no more than +`qos.max-sleepers` threads sleep at once and no request is delayed +longer than `qos.max-sleep-ms`; anything that would exceed either is let +through and counted in the `passed` stat. A class generating requests +far faster than the limiter can absorb will exceed its rate. If +`passed` badly outweighs `throttled`, raise `qos.max-sleepers` -- +accepting that the queue is then likelier to stall. + +**Classification is cached per pid for ten seconds.** A pid recycled +onto a different class inside that window is briefly misclassified. + +**Only reads and writes are governed.** Metadata operations are not, +being neither large nor slow enough to be worth the syscalls. + +**Direct access to a branch is invisible.** Anything reading or writing +`/mnt/disk1` rather than the pool never reaches mergerfs and cannot be +classified. Rebalance scripts and the like need the host's own +`ionice`. + + +## Supported platforms + +Linux only. `ioprio_set` and `/proc//cgroup` have no equivalent +elsewhere; on other platforms the option parses and does nothing. + + +## Performance impact + +A ruleset that cannot change anything is detected at parse time and +skipped entirely. Otherwise classification costs one small `/proc` read +per process per ten seconds, cached per worker thread, and only for the +fields some rule actually uses. Applying a class costs one +`ioprio_set` and one `setpriority` per *change*, not per request. diff --git a/mkdocs/mkdocs.yml b/mkdocs/mkdocs.yml index a8fc518c..ded08c75 100644 --- a/mkdocs/mkdocs.yml +++ b/mkdocs/mkdocs.yml @@ -82,6 +82,7 @@ nav: - config/threads.md - config/pin-threads.md - config/proxy-ioprio.md + - config/qos.md - config/link-cow.md - config/fuse-msg-size.md - config/follow-symlinks.md diff --git a/src/config.cpp b/src/config.cpp index 68539f77..bc6031f1 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -283,6 +283,8 @@ Config::Config() _map["remember"] = &_remember; _map["remember-nodes"] = &_remember_nodes; _map["qos"] = &qos; + _map["qos.distress-factor"] = &qos_distress_factor; + _map["qos.distress-ms"] = &qos_distress_ms; _map["qos.max-sleep-ms"] = &qos_max_sleep_ms; _map["qos.max-sleepers"] = &qos_max_sleepers; _map["qos.rules"] = &qos_rules; diff --git a/src/config.hpp b/src/config.hpp index cf1a20a6..6c4bfa9a 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -159,6 +159,8 @@ public: QoSRules qos_rules; QoSRuleSet qos_ruleset; QoSStats qos_stats; + QoSDistressMS qos_distress_ms; + QoSDistressFactor qos_distress_factor; QoSMaxSleepers qos_max_sleepers; QoSMaxSleepMS qos_max_sleep_ms; TFSRef read_thread_count; diff --git a/src/config_qos.cpp b/src/config_qos.cpp index 72d5091e..92f259b8 100644 --- a/src/config_qos.cpp +++ b/src/config_qos.cpp @@ -22,7 +22,10 @@ #include "qos.hpp" #include "syslog.hpp" +#include "fmt/core.h" + #include +#include QoS::QoS(const bool b_) @@ -130,7 +133,8 @@ QoSMaxSleepers::from_string(const std::string_view s_) rv = str::from(s_,&n); if(rv) return rv; - if(n < 0) + // -1 keeps the automatic sizing. + if(n < -1) return -EINVAL; qos::max_sleepers.store(n,std::memory_order_relaxed); @@ -162,3 +166,51 @@ QoSMaxSleepMS::from_string(const std::string_view s_) return 0; } + +std::string +QoSDistressMS::to_string(void) const +{ + return std::to_string(qos::distress_floor_ns.load(std::memory_order_relaxed) / + (1000 * 1000)); +} + +int +QoSDistressMS::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::distress_floor_ns.store(static_cast(n) * 1000 * 1000, + std::memory_order_relaxed); + + return 0; +} + +std::string +QoSDistressFactor::to_string(void) const +{ + return fmt::format("{:.2f}", + qos::distress_factor.load(std::memory_order_relaxed)); +} + +int +QoSDistressFactor::from_string(const std::string_view s_) +{ + const std::string str{s_}; + char *end = nullptr; + + errno = 0; + const double v = ::strtod(str.c_str(),&end); + if(errno || (end == str.c_str()) || (*end != '\0') || (v < 1.0)) + return -EINVAL; + + qos::distress_factor.store(v,std::memory_order_relaxed); + + return 0; +} diff --git a/src/config_qos.hpp b/src/config_qos.hpp index 0a58a51d..8eb5bfdf 100644 --- a/src/config_qos.hpp +++ b/src/config_qos.hpp @@ -71,6 +71,22 @@ public: int from_string(const std::string_view) final; }; +// qos.distress-ms= +class QoSDistressMS : public ToFromString +{ +public: + std::string to_string(void) const final; + int from_string(const std::string_view) final; +}; + +// qos.distress-factor= +class QoSDistressFactor : 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 { diff --git a/src/fuse_read.cpp b/src/fuse_read.cpp index e3d5209a..220ea896 100644 --- a/src/fuse_read.cpp +++ b/src/fuse_read.cpp @@ -67,19 +67,31 @@ 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; - if(fi->direct_io) - return ::_read_direct_io(fi->fd,buf_,size_,offset_); + // Classified after the file is known so path rules can see which + // file is being read, and so the rate is charged against the branch + // actually serving it. + qos::Apply q(ctx_,&fi->fusepath.native(),qos::Direction::READ); + qos::throttle(q,size_,fi->branch.path.native()); - return ::_read_cached(fi->fd,buf_,size_,offset_); + // Timed only for protected classes. How long a player's reads + // actually take is the signal the governor throttles everything + // else against, so it has to be measured around the real I/O rather + // than inferred. + const u64 t0 = qos::timing_start(q); + + const int rv = (fi->direct_io + ? ::_read_direct_io(fi->fd,buf_,size_,offset_) + : ::_read_cached(fi->fd,buf_,size_,offset_)); + + qos::timing_end(q,fi->branch.path.native(),t0); + + return rv; } int diff --git a/src/fuse_write.cpp b/src/fuse_write.cpp index e6916d85..0f3c34da 100644 --- a/src/fuse_write.cpp +++ b/src/fuse_write.cpp @@ -129,18 +129,11 @@ _move_and_pwriten(char const *buf_, static int -_write(const fuse_req_ctx_t *ctx_, - const fuse_file_info_t *ffi_, - const char *buf_, - const size_t count_, - const off_t offset_) +_write(FileInfo *fi, + const char *buf_, + const size_t count_, + const off_t offset_) { - FileInfo *fi; - - fi = state.get_fi(ctx_,ffi_->fh); - if(not fi) - return -EBADF; - // Concurrent writes can only happen if: // 1) writeback-cache is enabled and using page caching // 2) parallel_direct_writes is enabled and file has @@ -203,14 +196,26 @@ FUSE::write(const fuse_req_ctx_t *ctx_, off_t offset_) { ioprio::SetFrom iop(ctx_->pid); - qos::Apply q(ctx_); + FileInfo *fi; + + fi = state.get_fi(ctx_,ffi_->fh); + if(not fi) + return -EBADF; + + qos::Apply q(ctx_,&fi->fusepath.native(),qos::Direction::WRITE); // 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_); + qos::throttle(q,count_,fi->branch.path.native()); - return ::_write(ctx_,ffi_,buf_,count_,offset_); + const u64 t0 = qos::timing_start(q); + + const int rv = ::_write(fi,buf_,count_,offset_); + + qos::timing_end(q,fi->branch.path.native(),t0); + + return rv; } int diff --git a/src/procfs.cpp b/src/procfs.cpp index 7d2bca9b..0e819647 100644 --- a/src/procfs.cpp +++ b/src/procfs.cpp @@ -135,3 +135,42 @@ procfs::get_cgroup(const int pid_) return buf.data(); } + +std::string +procfs::get_cmdline(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_cmdline called before procfs::init()"); + + frv = fmt::format_to_n(buf.data(),buf.size()-1,"{}/cmdline",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 {}; + + // Arguments are NUL separated; joining them with spaces is what + // makes a glob like "*/Transcode/Detection/*" work against the + // whole invocation. + for(int i = 0; i < rv; i++) + { + if(buf[i] == '\0') + buf[i] = ' '; + } + + if(buf[rv-1] == ' ') + rv--; + buf[rv] = '\0'; + + return buf.data(); +} diff --git a/src/procfs.hpp b/src/procfs.hpp index b3907f61..9a5ceb0e 100644 --- a/src/procfs.hpp +++ b/src/procfs.hpp @@ -31,4 +31,10 @@ namespace procfs // 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); + + // /proc//cmdline with the NUL separators turned into spaces. + // Truncated to a bounded length: the interesting part of a media + // server's command line is the input and output paths, and a full + // ffmpeg invocation can run to kilobytes. + std::string get_cmdline(const int pid); } diff --git a/src/qos.cpp b/src/qos.cpp index 97e6f707..04cc0a80 100644 --- a/src/qos.cpp +++ b/src/qos.cpp @@ -18,6 +18,7 @@ #include "qos.hpp" +#include "config.hpp" #include "ioprio.hpp" #include "procfs.hpp" @@ -27,12 +28,19 @@ #include #include +#include +#include #include +#include #include #include +#include +#include using qos::Class; +using qos::Direction; using qos::RuleSet; +using qos::Subject; namespace qos { @@ -41,8 +49,15 @@ namespace qos // 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}; + // -1 means "size it from the process thread pool". A fixed small + // number under-enforces badly on a fast pool, and a large one risks + // the head-of-line stall described in qos.hpp; half the pool keeps + // the other half free to serve whoever is being protected. + std::atomic max_sleepers{-1}; std::atomic max_sleep_ns{50ULL * 1000 * 1000}; + + std::atomic distress_floor_ns{50ULL * 1000 * 1000}; + std::atomic distress_factor{3.0}; } namespace @@ -64,6 +79,44 @@ namespace // ended, throttling traffic that did nothing wrong. constexpr u64 MAX_DEBT_NS = 2 * NS_PER_SEC; + // How recently a protected class must have issued I/O on a resource + // for that resource to count as contended. Past this, yielding + // classes go back to full speed. + constexpr u64 CONTENTION_WINDOW_NS = 2 * NS_PER_SEC; + + // A protected request is "in distress" when the smoothed service + // time exceeds this multiple of the quietest time the same resource + // has managed -- but never below the absolute floor, so ordinary + // jitter on a fast device does not trip the loop. + // Multiplicative increase on distress, additive decrease per second + // of calm. Deliberately asymmetric: a stutter has already been heard + // by the time it is measured, so back off hard and return gently. + constexpr double PRESSURE_UP_FACTOR = 1.5; + constexpr double PRESSURE_UP_STEP = 0.10; + constexpr double PRESSURE_DOWN_PER_SEC = 0.20; + + std::mutex g_gov_mutex; + + // Governor state is keyed by resource and deliberately outlives + // ruleset reloads: what a disk is doing does not change because + // somebody edited a rules file. + std::unordered_map> g_governors; + + qos::Governor * + _governor_for(const std::string &resource_) + { + std::lock_guard lk(g_gov_mutex); + + auto i = g_governors.find(resource_); + if(i != g_governors.end()) + return i->second.get(); + + auto [it,inserted] = g_governors.emplace(resource_, + std::make_unique()); + + return it->second.get(); + } + std::mutex g_mutex; std::atomic g_generation{1}; std::atomic g_sleepers{0}; @@ -82,8 +135,10 @@ namespace struct CacheEntry { u32 pid = 0; + u64 key = 0; // path hash folded with the direction u64 expire = 0; const Class *cls = nullptr; + bool valid = false; }; // Per worker thread state. Holding the RuleSet::Ptr here, rather @@ -105,6 +160,41 @@ namespace thread_local TLS tls; + // Resolves the -1 "auto" sentinel against the running thread pool. + int + _sleeper_limit() + { + const int configured = qos::max_sleepers.load(std::memory_order_relaxed); + + if(configured >= 0) + return configured; + + // Resolved once: the thread count is fixed when the mount comes + // up, and this sits on the throttle path where building a string + // per request would be absurd. + static const int half = + []() + { + // The config holds the *requested* value, where <= 0 means + // "decide for me" -- reading it raw yields 0 and a ceiling of + // one sleeper. When it has not been pinned, mirror libfuse's + // own default so the ceiling tracks the pool that actually + // exists. + int threads = std::atoi(cfg.process_thread_count.to_string().c_str()); + + if(threads <= 0) + { + const int nproc = static_cast(std::thread::hardware_concurrency()); + + threads = std::min(8,std::max(2,nproc - 2)); + } + + return std::max(1,threads / 2); + }(); + + return half; + } + u64 _now_ns() { @@ -135,27 +225,48 @@ namespace tls.next = 0; } - const Class * - _lookup(const u32 pid_, - const u64 now_) + // The cache is keyed by the whole subject, not just the pid: with + // path rules in play the same process reading two different folders + // can land in two different classes. + bool + _lookup(const u32 pid_, + const u64 key_, + const u64 now_, + const Class **cls_) { for(const auto &e : tls.cache) { - if((e.pid == pid_) && (now_ < e.expire)) - return e.cls; + if(e.valid && (e.pid == pid_) && (e.key == key_) && (now_ < e.expire)) + { + *cls_ = e.cls; + return true; + } } - return nullptr; + return false; } void _remember(const u32 pid_, + const u64 key_, const u64 now_, const Class *cls_) { - tls.cache[tls.next] = CacheEntry{pid_,now_ + CACHE_TTL_NS,cls_}; + tls.cache[tls.next] = CacheEntry{pid_,key_,now_ + CACHE_TTL_NS,cls_,true}; tls.next = ((tls.next + 1) % CACHE_SIZE); } + + u64 + _subject_key(const std::string *path_, + const qos::Direction dir_) + { + u64 key = ((dir_ == qos::Direction::WRITE) ? 1u : 0u); + + if(path_ != nullptr) + key ^= (std::hash{}(*path_) << 1); + + return key; + } } void @@ -215,56 +326,259 @@ qos::load_file(const std::string &path_, return 0; } -void -qos::throttle(const Class *cls_, - const u64 bytes_) +u64 +qos::timing_start(const Apply &apply_) { - if(cls_ == nullptr) + const Class *cls = apply_.cls(); + + // Only protected classes drive the loop, so everything else pays + // nothing -- not even a clock read. + if((cls == nullptr) || !cls->protect) + return 0; + + return ::_now_ns(); +} + +void +qos::timing_end(const Apply &apply_, + const std::string &resource_, + const u64 started_) +{ + if(started_ == 0) return; - cls_->requests.fetch_add(1,std::memory_order_relaxed); - cls_->bytes.fetch_add(bytes_,std::memory_order_relaxed); + const u64 now = ::_now_ns(); + const u64 latency = ((now > started_) ? (now - started_) : 0); - if(cls_->rate == 0) + qos::Governor *g = ::_governor_for(resource_); + + std::lock_guard lk(g->mutex); + + g->protected_at = now; + + // Backing off only helps if there is something to back off. A read + // that is slow on an otherwise idle disk is just a slow disk, and + // throttling an absent competitor would not speed it up. + const bool contended = (g->yielding_at && + ((now - g->yielding_at) <= CONTENTION_WINDOW_NS)); + + if(g->latency_ewma == 0) + g->latency_ewma = latency; + else + g->latency_ewma = (((g->latency_ewma * 7) + latency) / 8); + + // The baseline is what this resource does when nothing is fighting + // it. It follows a new low immediately but drifts back up only + // slowly, so one unusually fast (cached) read cannot permanently + // convince the governor that every later read is in distress. + if(g->latency_base == 0) + g->latency_base = g->latency_ewma; + else if(g->latency_ewma < g->latency_base) + g->latency_base = g->latency_ewma; + else + g->latency_base += ((g->latency_ewma - g->latency_base) / 1024); + + const u64 threshold = + std::max(static_cast(g->latency_base * + qos::distress_factor.load(std::memory_order_relaxed)), + qos::distress_floor_ns.load(std::memory_order_relaxed)); + + const u64 elapsed = ((g->updated_at && (now > g->updated_at)) + ? (now - g->updated_at) + : 0); + + if(contended && (g->latency_ewma > threshold)) + { + g->pressure = std::min(1.0, + (g->pressure * PRESSURE_UP_FACTOR) + PRESSURE_UP_STEP); + g->distress_events++; + } + else + { + const double secs = (static_cast(elapsed) / NS_PER_SEC); + g->pressure = std::max(0.0, + g->pressure - (PRESSURE_DOWN_PER_SEC * secs)); + } + + g->updated_at = now; +} + +void +qos::note_yielding(const std::string &resource_) +{ + qos::Governor *g = ::_governor_for(resource_); + + const u64 now = ::_now_ns(); + + std::lock_guard lk(g->mutex); + + g->yielding_at = now; +} + +double +qos::pressure(const std::string &resource_) +{ + qos::Governor *g = ::_governor_for(resource_); + + const u64 now = ::_now_ns(); + + std::lock_guard lk(g->mutex); + + // Nobody has been streaming off this resource lately, so there is + // nothing to protect and nothing to yield to. + if((g->protected_at == 0) || + ((now - g->protected_at) > CONTENTION_WINDOW_NS)) + { + g->pressure = 0.0; + return 0.0; + } + + // Decay for time passed since the last protected sample so pressure + // keeps falling even while the only traffic is the bulk class + // reading this value. + if(g->updated_at && (now > g->updated_at)) + { + const double secs = (static_cast(now - g->updated_at) / NS_PER_SEC); + const double decayed = (g->pressure - (PRESSURE_DOWN_PER_SEC * secs)); + + if(decayed < g->pressure) + { + g->pressure = std::max(0.0,decayed); + g->updated_at = now; + } + } + + return g->pressure; +} + +void +qos::throttle(const Apply &apply_, + const u64 bytes_, + const std::string &resource_) +{ + const Class *cls = apply_.cls(); + + if(cls == nullptr) return; + cls->requests.fetch_add(1,std::memory_order_relaxed); + cls->bytes.fetch_add(bytes_,std::memory_order_relaxed); + + // Protected traffic and critical dependencies are never delayed, + // whatever else the ruleset says about them. Checked before any + // rate is even resolved so a `rate=` accidentally left on such a + // class cannot slow the stream waiting behind it. + if(cls->immune()) + return; + + // An adaptive class normally carries no static rate at all -- its + // whole allowance comes from the governor -- so it must not be + // dismissed here for looking unlimited. + if(!cls->limited() && !cls->adaptive()) + return; + + const RuleSet *rs = apply_.ruleset(); + if(rs == nullptr) + return; + + // A percentage only becomes a number once the serving branch is + // known, so the rate is resolved per request rather than at parse + // time. + u64 rate = rs->rate_for(cls,resource_); + + if(cls->adaptive()) + { + // Tell the governor there is traffic here that *could* give way. + // Without this it cannot distinguish "playback is slow because + // downloads are hammering the disk" from "playback is slow + // because the disk is slow", and only the first is worth + // reacting to. + qos::note_yielding(resource_); + + const double p = qos::pressure(resource_); + + if(p <= 0.0) + { + // Nothing protected is reading this resource, so a yielding + // class runs at whatever it was configured for -- which for + // downloads is normally no limit at all. + if(rate == 0) + return; + } + else + { + // Scaling needs a number to scale. An otherwise unlimited + // class is measured against the resource's capacity; with no + // capacity declared there is nothing to compute against and + // it stays unlimited. + const u64 base = (rate ? rate : rs->capacity(resource_)); + + if(base == 0) + return; + + const double share = (1.0 - ((p * cls->yield) / 100.0)); + u64 scaled = static_cast(base * std::max(0.0,share)); + + // Yielding must never mean stopping: a floored class keeps + // making progress, just slowly. + const u64 floor = rs->floor_for(cls,resource_); + if(scaled < floor) + scaled = floor; + if(scaled == 0) + scaled = 1; + + rate = scaled; + } + } + + if(rate == 0) + return; + + const u64 burst = (cls->burst ? cls->burst : rate); + + qos::Bucket *bucket = cls->bucket_for(resource_); + u64 sleep_ns; { - std::lock_guard lk(cls_->bucket_mutex); + std::lock_guard lk(bucket->mutex); const u64 now = ::_now_ns(); - if(cls_->refilled == 0) - cls_->refilled = now; + if(!bucket->primed) + { + bucket->tokens = burst; + bucket->refilled = now; + bucket->primed = true; + } - if(now > cls_->refilled) + if(now > bucket->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 elapsed = (now - bucket->refilled); const unsigned __int128 added = - ((elapsed * static_cast(cls_->rate)) / NS_PER_SEC); + ((elapsed * static_cast(rate)) / NS_PER_SEC); - if(added >= cls_->burst) - cls_->tokens = cls_->burst; + if(added >= burst) + bucket->tokens = burst; else - cls_->tokens = std::min(cls_->burst, - cls_->tokens + static_cast(added)); + bucket->tokens = std::min(burst, + bucket->tokens + static_cast(added)); - cls_->refilled = now; + bucket->refilled = now; } - if(cls_->tokens >= bytes_) + if(bucket->tokens >= bytes_) { - cls_->tokens -= bytes_; + bucket->tokens -= bytes_; return; } - const u64 deficit = (bytes_ - cls_->tokens); + const u64 deficit = (bytes_ - bucket->tokens); - cls_->tokens = 0; + bucket->tokens = 0; // Carry the debt by pushing the refill point into the future // rather than forgiving it. Tokens accrue in real time, so a @@ -276,30 +590,30 @@ qos::throttle(const Class *cls_, // 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); + NS_PER_SEC) / rate); - cls_->refilled += need_ns; - if(cls_->refilled > (now + MAX_DEBT_NS)) - cls_->refilled = (now + MAX_DEBT_NS); + bucket->refilled += need_ns; + if(bucket->refilled > (now + MAX_DEBT_NS)) + bucket->refilled = (now + MAX_DEBT_NS); - sleep_ns = ((cls_->refilled > now) ? (cls_->refilled - now) : 0); + sleep_ns = ((bucket->refilled > now) ? (bucket->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); + 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)) + if(sleepers > ::_sleeper_limit()) { g_sleepers.fetch_sub(1,std::memory_order_relaxed); - cls_->passed.fetch_add(1,std::memory_order_relaxed); + cls->passed.fetch_add(1,std::memory_order_relaxed); return; } @@ -310,12 +624,14 @@ qos::throttle(const Class *cls_, 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); + 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_) +qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_, + const std::string *fusepath_, + const Direction dir_) { ::_refresh_ruleset(); @@ -323,10 +639,13 @@ qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_) if((rs == nullptr) || rs->inert()) return; - const u64 now = ::_now_ns(); + _rs = rs; - const Class *cls = ::_lookup(ctx_->pid,now); - if(cls == nullptr) + const u64 now = ::_now_ns(); + const u64 key = ::_subject_key((rs->needs_path() ? fusepath_ : nullptr),dir_); + + const Class *cls; + if(!::_lookup(ctx_->pid,key,now,&cls)) { // Only the fields some rule actually tests are read, so a // ruleset matching on cgroup alone costs one file read per @@ -337,9 +656,22 @@ qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_) const std::string comm = (rs->needs_comm() ? procfs::get_name(ctx_->pid) : std::string{}); + const std::string cmdline = (rs->needs_cmdline() + ? procfs::get_cmdline(ctx_->pid) + : std::string{}); - cls = rs->classify(cgroup,comm,ctx_->uid,ctx_->gid); - ::_remember(ctx_->pid,now,cls); + Subject subject; + subject.cgroup = &cgroup; + subject.comm = &comm; + subject.cmdline = &cmdline; + subject.path = fusepath_; + subject.uid = ctx_->uid; + subject.gid = ctx_->gid; + subject.dir = dir_; + + cls = rs->classify(subject); + + ::_remember(ctx_->pid,key,now,cls); } if(cls == nullptr) @@ -356,7 +688,7 @@ qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_) if((cls->nice != qos::UNSET) && (cls->nice != tls.applied_nice)) { errno = 0; - if(::setpriority(PRIO_PROCESS,0,cls->nice) == 0 || errno == 0) + if((::setpriority(PRIO_PROCESS,0,cls->nice) == 0) || (errno == 0)) tls.applied_nice = cls->nice; } } @@ -382,10 +714,32 @@ qos::stats() "max-sleep-ms={}\n", (qos::enabled() ? "true" : "false"), g_sleepers.load(std::memory_order_relaxed), - qos::max_sleepers.load(std::memory_order_relaxed), + ::_sleeper_limit(), (qos::max_sleep_ns.load(std::memory_order_relaxed) / (1000 * 1000))); + { + std::lock_guard lk(g_gov_mutex); + const u64 now = ::_now_ns(); + + for(const auto &[resource,g] : g_governors) + { + std::lock_guard glk(g->mutex); + + const bool contended = (g->protected_at && + ((now - g->protected_at) <= CONTENTION_WINDOW_NS)); + + out += fmt::format("resource {}: contended={} pressure={:.2f} " + "latency-ms={:.1f} baseline-ms={:.1f} distress={}\n", + resource, + (contended ? "yes" : "no"), + (contended ? g->pressure : 0.0), + (static_cast(g->latency_ewma) / 1000000.0), + (static_cast(g->latency_base) / 1000000.0), + g->distress_events); + } + } + for(const auto &c : rs->classes()) { out += fmt::format("{}: requests={} bytes={} throttled={} " diff --git a/src/qos.hpp b/src/qos.hpp index 9ba47a91..d4f83d05 100644 --- a/src/qos.hpp +++ b/src/qos.hpp @@ -88,16 +88,60 @@ namespace qos extern std::atomic max_sleepers; extern std::atomic max_sleep_ns; + // What counts as a protected request being in distress: its smoothed + // service time must exceed `distress_factor` times the quietest time + // that resource has managed, and also exceed `distress_floor_ns`. + // + // The floor is what stops ordinary jitter tripping the loop, and it + // is entirely device dependent -- 50ms is a stall on a spinning disk + // and unreachable on NVMe. A flash pool wants single-digit + // milliseconds or the governor will never engage. + extern std::atomic distress_floor_ns; + extern std::atomic distress_factor; + + class Apply; + // 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); + // Feedback loop. + // + // A protected class's service time is the control signal. When reads + // for a player start taking materially longer than that same disk + // manages when it is quiet, pressure on the resource rises and every + // yielding class's allowance shrinks in proportion to its `yield` + // rank -- downloads first and hardest, a player's own background + // work more gently, playback not at all. When the latency recovers, + // or playback stops altogether, pressure decays and the allowances + // return to full. + // + // Returns 0 for a request whose class is not protected, in which + // case timing_end does nothing. Keeping the clock read out of the + // unprotected path is why this is split in two. + u64 timing_start(const Apply &); + void timing_end(const Apply &, const std::string &resource, const u64 started); + + // Current backoff for a resource, 0.0 to 1.0. Zero when no protected + // class has touched it recently, which is what lets bulk traffic run + // flat out on a disk nobody is streaming from. + double pressure(const std::string &resource); + + // Records that a class willing to yield has just issued I/O against + // a resource, which is what makes it count as contended. + void note_yielding(const std::string &resource); + + // Charges `bytes` to the class and, if that class is over its rate + // for `resource`, sleeps for as long as the bounds above permit. + // Called with the size the caller asked for, before the I/O is + // issued. + // + // `resource` is the branch serving the request, so a rate is a + // per-device allowance: a class limited to 10% does not have to + // share one budget across seven disks. + void throttle(const Apply &, const u64 bytes, const std::string &resource); // Classifies the calling process and applies its class to this // thread. The thread keeps those settings until another request @@ -109,19 +153,25 @@ namespace qos public: [[gnu::always_inline]] inline - Apply(const fuse_req_ctx_t *ctx_) + Apply(const fuse_req_ctx_t *ctx_, + const std::string *fusepath_, + const Direction dir_) { if(qos::enabled()) - _slow_apply(ctx_); + _slow_apply(ctx_,fusepath_,dir_); } // nullptr when QoS is off or the ruleset cannot change anything. - const Class *cls() const { return _cls; } + const Class *cls() const { return _cls; } + const RuleSet *ruleset() const { return _rs; } private: - void _slow_apply(const fuse_req_ctx_t *); + void _slow_apply(const fuse_req_ctx_t *, + const std::string *, + const Direction); private: - const Class *_cls = nullptr; + const Class *_cls = nullptr; + const RuleSet *_rs = nullptr; }; } diff --git a/src/qos_class.cpp b/src/qos_class.cpp new file mode 100644 index 00000000..6d6773e1 --- /dev/null +++ b/src/qos_class.cpp @@ -0,0 +1,37 @@ +/* + 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_class.hpp" + + +qos::Bucket * +qos::Class::bucket_for(const std::string &resource_) const +{ + std::lock_guard lk(buckets_mutex); + + auto i = buckets.find(resource_); + if(i != buckets.end()) + return i->second.get(); + + // unique_ptr rather than the value type because a Bucket holds a + // mutex: rehashing the map must not move one out from under a + // thread waiting on it. + auto [it,inserted] = buckets.emplace(resource_,std::make_unique()); + + return it->second.get(); +} diff --git a/src/qos_class.hpp b/src/qos_class.hpp index 03f82622..596dfef0 100644 --- a/src/qos_class.hpp +++ b/src/qos_class.hpp @@ -23,6 +23,8 @@ #include #include #include +#include +#include namespace qos @@ -72,14 +74,61 @@ namespace qos int from_string(const std::string_view, int *value); } + // One token bucket. A class owns one of these per resource it is + // limited against -- normally one per branch, so a rate is a + // per-device rate rather than a pool-wide one. Two players reading + // two different disks do not compete for the same allowance. + struct Bucket + { + std::mutex mutex; + u64 tokens = 0; + u64 refilled = 0; + bool primed = false; + }; + + // The feedback state for one resource (one branch/spindle). + // + // This is what makes the policy adaptive rather than a fixed cap: + // bulk traffic runs unrestricted on a disk nobody is streaming from, + // and only gives way -- gradually, and never to a standstill -- once + // a protected class on that same disk starts to suffer. + struct Governor + { + std::mutex mutex; + + // When a protected class last issued I/O here. Bulk classes read + // this to answer "is anyone playing off this spindle right now?" + u64 protected_at = 0; + + // When a yielding class last issued I/O here. Pressure only rises + // while both this and protected_at are recent -- that is the + // definition of contention. + u64 yielding_at = 0; + + // Exponentially weighted mean service time of protected requests, + // and the quietest mean seen so far, which stands in for "what + // this disk does when nothing is fighting it". + u64 latency_ewma = 0; + u64 latency_base = 0; + + // 0.0 == no backoff, 1.0 == full backoff. Moved multiplicatively + // up on distress and additively down on recovery, so it reacts + // fast to a stutter and returns slowly enough not to oscillate. + double pressure = 0.0; + + u64 updated_at = 0; + + // Counters for qos.stats. + u64 distress_events = 0; + }; + // A QoS class is a named bundle of scheduling parameters plus the - // token bucket and counters belonging to every request matched to + // token buckets 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. + // The descriptive members are immutable once a ruleset is + // published, which is what lets the classifier hand out bare + // pointers to it. Everything mutable is either atomic or guarded. class Class { public: @@ -94,19 +143,58 @@ namespace qos 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. + // A rate is either absolute (`rate` bytes/sec) or a share of the + // serving resource's measured capacity (`pct` percent). Both zero + // means unlimited. u64 rate = 0; + u32 pct = 0; + // Bucket depth. Zero means "one second of whatever the rate + // resolves to", which cannot be precomputed for a percentage. u64 burst = 0; + // Latency of this class's requests is the signal the governor + // controls against, and it is never itself throttled. + bool protect = false; + + // Never delayed, and never a control signal. + // + // This exists for the services playback *synchronously waits on* + // rather than playback itself -- Plex's EasyAudioEncoder is the + // canonical case: a transcode of EAC3/TrueHD/DTS audio writes to + // EAE and blocks until it answers. Slowing such a helper slows the + // stream that is waiting on it, so the "bulk" it appears to be + // doing must be left alone. Getting this wrong is a priority + // inversion: the machinery meant to protect playback stalls it. + bool critical = false; + + // True when this class must never be delayed for any reason. + bool immune() const { return (protect || critical); } + + // How strongly this class gives way as pressure rises, 0-100. + // 0 never yields; 100 yields its whole allowance at full + // pressure. This is the priority ladder: downloads yield hardest, + // a player's background work yields some, playback yields none. + u32 yield = 0; + + // Never back off below this, so a yielding class is slowed rather + // than stopped. Either absolute bytes/sec or a percentage of the + // resource's capacity. + u64 floor = 0; + u32 floor_pct = 0; + + bool limited() const { return ((rate != 0) || (pct != 0)); } + bool adaptive() const { return (yield != 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}; + // Buckets are created on first use and keyed by resource name + // (a branch path, or "pool" when the branch is unknown). Seven + // branches means seven entries, so a plain map under a mutex is + // cheaper than anything cleverer. + mutable std::mutex buckets_mutex; + mutable std::unordered_map> buckets; + + Bucket *bucket_for(const std::string &resource) const; public: // Counters, reported via the qos.stats config key. diff --git a/src/qos_rules.cpp b/src/qos_rules.cpp index bbd31fac..ce3e528d 100644 --- a/src/qos_rules.cpp +++ b/src/qos_rules.cpp @@ -28,10 +28,13 @@ #include using qos::Class; +using qos::Condition; +using qos::Direction; using qos::Field; -using qos::Op; +using qos::MatchOp; using qos::Rule; using qos::RuleSet; +using qos::Subject; std::string @@ -144,28 +147,56 @@ qos::parse_size(const std::string_view s_, } bool -Rule::matches(const std::string &cgroup_, - const std::string &comm_, - const u32 uid_, - const u32 gid_) const +qos::Condition::matches(const Subject &s_) const { + const std::string *subject; + switch(field) { case Field::UID: - return (uid_ == number); + return (s_.uid == number); case Field::GID: - return (gid_ == number); + return (s_.gid == number); + case Field::OP: + return (s_.dir == dir); case Field::CGROUP: - case Field::COMM: + subject = s_.cgroup; break; + case Field::COMM: + subject = s_.comm; + break; + case Field::CMDLINE: + subject = s_.cmdline; + break; + case Field::PATH: + subject = s_.path; + break; + default: + return false; } - const std::string &subject = ((field == Field::CGROUP) ? cgroup_ : comm_); + if(subject == nullptr) + return false; - if(op == Op::EQ) - return (subject == pattern); + if(op == MatchOp::EQ) + return (*subject == pattern); - return (::fnmatch(pattern.c_str(),subject.c_str(),0) == 0); + // FNM_PATHNAME is deliberately not set: a rule reading + // "path ~ /TV/*" is expected to cover everything beneath /TV, which + // is what an administrator writing a folder rule means. + return (::fnmatch(pattern.c_str(),subject->c_str(),0) == 0); +} + +bool +qos::Rule::matches(const Subject &s_) const +{ + for(const auto &cond : conditions) + { + if(!cond.matches(s_)) + return false; + } + + return true; } const Class * @@ -181,20 +212,67 @@ RuleSet::find_class(const std::string_view name_) const } const Class * -RuleSet::classify(const std::string &cgroup_, - const std::string &comm_, - const u32 uid_, - const u32 gid_) const +RuleSet::classify(const Subject &s_) const { for(const auto &rule : _rules) { - if(rule.matches(cgroup_,comm_,uid_,gid_)) + if(rule.matches(s_)) return rule.cls; } return _default; } +u64 +RuleSet::capacity(const std::string &resource_) const +{ + auto i = _capacity.find(resource_); + + if(i != _capacity.end()) + return i->second; + + return _capacity_default; +} + +u64 +RuleSet::floor_for(const Class *cls_, + const std::string &resource_) const +{ + if(cls_->floor) + return cls_->floor; + + if(cls_->floor_pct == 0) + return 0; + + const u64 cap = capacity(resource_); + if(cap == 0) + return 0; + + return ((cap * cls_->floor_pct) / 100); +} + +u64 +RuleSet::rate_for(const Class *cls_, + const std::string &resource_) const +{ + if(cls_->rate) + return cls_->rate; + + if(cls_->pct == 0) + return 0; + + const u64 cap = capacity(resource_); + + // No capacity known for this resource means a percentage cannot be + // turned into a number. Running unlimited is the safe reading: a + // guessed ceiling would throttle traffic against a figure nobody + // supplied. + if(cap == 0) + return 0; + + return ((cap * cls_->pct) / 100); +} + RuleSet::Ptr RuleSet::make_default() { @@ -209,14 +287,57 @@ RuleSet::make_default() namespace { + // Whitespace separated, except that a token may be wrapped in single + // or double quotes to keep spaces in it. Necessary rather than + // decorative: the process names this has to match include + // "Plex Transcoder" and "Plex Media Scanner", and a command line is + // nothing but spaces. std::vector tokenize(const std::string &line_) { std::vector tokens; - std::istringstream iss(line_); std::string tok; + bool in_tok = false; + char quote = '\0'; - while(iss >> tok) + for(std::size_t i = 0; i < line_.size(); i++) + { + const char c = line_[i]; + + if(quote) + { + if(c == quote) + quote = '\0'; + else + tok += c; + continue; + } + + if((c == '"') || (c == '\'')) + { + // A quote starts a token even when empty, so that '' is a + // deliberate empty pattern rather than nothing at all. + quote = c; + in_tok = true; + continue; + } + + if(std::isspace(static_cast(c))) + { + if(in_tok) + { + tokens.push_back(tok); + tok.clear(); + in_tok = false; + } + continue; + } + + tok += c; + in_tok = true; + } + + if(in_tok) tokens.push_back(tok); return tokens; @@ -234,6 +355,12 @@ namespace *field_ = Field::UID; else if(s_ == "gid") *field_ = Field::GID; + else if(s_ == "cmdline") + *field_ = Field::CMDLINE; + else if(s_ == "path") + *field_ = Field::PATH; + else if(s_ == "op") + *field_ = Field::OP; else return -EINVAL; @@ -281,6 +408,19 @@ RuleSet::parse(const std::string_view text_, for(std::size_t i = 2; i < tokens.size(); i++) { + // `protect` is a bare flag rather than key=value. + if(tokens[i] == "protect") + { + cls->protect = true; + continue; + } + + if(tokens[i] == "critical") + { + cls->critical = true; + continue; + } + const std::size_t eq = tokens[i].find('='); if(eq == std::string::npos) return fail(fmt::format("expected key=value, got '{}'",tokens[i])); @@ -302,14 +442,51 @@ RuleSet::parse(const std::string_view text_, } else if(key == "rate") { - if(qos::parse_size(val,&cls->rate)) - return fail(fmt::format("invalid rate '{}'",val)); + if(!val.empty() && (val.back() == '%')) + { + const std::string num = val.substr(0,val.size() - 1); + char *end = nullptr; + const unsigned long pct = ::strtoul(num.c_str(),&end,10); + if((end == num.c_str()) || (*end != '\0') || + (pct == 0) || (pct > 100)) + return fail(fmt::format("percentage must be 1-100: '{}'",val)); + cls->pct = static_cast(pct); + } + else 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 if(key == "yield") + { + char *end = nullptr; + const unsigned long y = ::strtoul(val.c_str(),&end,10); + if((end == val.c_str()) || (*end != '\0') || (y > 100)) + return fail(fmt::format("yield must be 0-100: '{}'",val)); + cls->yield = static_cast(y); + } + else if(key == "floor") + { + if(!val.empty() && (val.back() == '%')) + { + const std::string num = val.substr(0,val.size() - 1); + char *end = nullptr; + const unsigned long pct = ::strtoul(num.c_str(),&end,10); + if((end == num.c_str()) || (*end != '\0') || + (pct == 0) || (pct > 100)) + return fail(fmt::format("floor percentage must be 1-100: '{}'",val)); + cls->floor_pct = static_cast(pct); + } + else if(qos::parse_size(val,&cls->floor)) + { + return fail(fmt::format("invalid floor '{}'",val)); + } + } else { return fail(fmt::format("unknown class key '{}'",key)); @@ -318,63 +495,127 @@ RuleSet::parse(const std::string_view text_, // 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) + // without letting an idle class bank unlimited credit. For + // a percentage rate the depth can only be resolved once the + // serving resource is known, so it is left at zero here. + if(cls->rate && (cls->burst < cls->rate)) cls->burst = cls->rate; - cls->tokens = cls->burst; + if(cls->protect) + rs->_has_protected = true; if((cls->ioprio != qos::UNSET) || (cls->nice != qos::UNSET) || - (cls->rate != 0)) + cls->limited() || + cls->adaptive() || + cls->protect || + cls->critical) rs->_inert = false; rs->_classes.push_back(std::move(cls)); } + else if(tokens[0] == "capacity") + { + if(tokens.size() != 3) + return fail("expected: capacity "); + + u64 rate; + if(qos::parse_size(tokens[2],&rate)) + return fail(fmt::format("invalid capacity '{}'",tokens[2])); + + if(tokens[1] == "default") + rs->_capacity_default = rate; + else + rs->_capacity[tokens[1]] = rate; + } 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{}; + std::size_t i = 1; - 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)) + // Conditions are (field, operator, pattern) triples running + // up to the arrow. Every one must match for the rule to + // fire. + while((i < tokens.size()) && (tokens[i] != "->")) { - if(rule.op != Op::EQ) - return fail("uid/gid only support the '=' operator"); + if((i + 2) >= tokens.size()) + return fail("expected: match <=|~> ... -> "); - 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); + Condition cond{}; + + if(::parse_field(tokens[i],&cond.field)) + return fail(fmt::format("unknown field '{}'",tokens[i])); + + if(tokens[i+1] == "=") + cond.op = MatchOp::EQ; + else if(tokens[i+1] == "~") + cond.op = MatchOp::GLOB; + else + return fail(fmt::format("unknown operator '{}'",tokens[i+1])); + + cond.pattern = tokens[i+2]; + + switch(cond.field) + { + case Field::UID: + case Field::GID: + { + if(cond.op != MatchOp::EQ) + return fail("uid/gid only support the '=' operator"); + + char *end = nullptr; + const unsigned long n = ::strtoul(cond.pattern.c_str(),&end,10); + if((end == cond.pattern.c_str()) || (*end != '\0')) + return fail(fmt::format("invalid uid/gid '{}'",cond.pattern)); + cond.number = static_cast(n); + } + break; + + case Field::OP: + if(cond.op != MatchOp::EQ) + return fail("op only supports the '=' operator"); + if(cond.pattern == "read") + cond.dir = Direction::READ; + else if(cond.pattern == "write") + cond.dir = Direction::WRITE; + else + return fail(fmt::format("op must be read or write, got '{}'", + cond.pattern)); + break; + + case Field::CGROUP: + rs->_needs_cgroup = true; + break; + + case Field::COMM: + rs->_needs_comm = true; + break; + + case Field::CMDLINE: + rs->_needs_cmdline = true; + break; + + case Field::PATH: + rs->_needs_path = true; + break; + } + + rule.conditions.push_back(std::move(cond)); + + i += 3; } - rule.pattern = tokens[3]; + if(rule.conditions.empty()) + return fail("match needs at least one condition"); + if((i >= tokens.size()) || (tokens[i] != "->")) + return fail("expected '->' after the last condition"); + if((i + 2) != tokens.size()) + return fail("expected exactly one class name after '->'"); - rule.cls = rs->find_class(tokens[5]); + rule.cls = rs->find_class(tokens[i+1]); 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; + "defined before use)",tokens[i+1])); rs->_rules.push_back(std::move(rule)); } @@ -407,8 +648,13 @@ RuleSet::parse(const std::string_view text_, } } - if(rs->_rules.empty() && rs->_default->ioprio == qos::UNSET && - rs->_default->nice == qos::UNSET && rs->_default->rate == 0) + if(rs->_rules.empty() && + (rs->_default->ioprio == qos::UNSET) && + (rs->_default->nice == qos::UNSET) && + !rs->_default->limited() && + !rs->_default->adaptive() && + !rs->_default->protect && + !rs->_default->critical) rs->_inert = true; return rs; @@ -419,32 +665,52 @@ RuleSet::to_string() const { std::string s; + for(const auto &[resource,rate] : _capacity) + s += fmt::format("capacity {} {}\n",resource,rate); + if(_capacity_default) + s += fmt::format("capacity default {}\n",_capacity_default); + for(const auto &c : _classes) { - s += fmt::format("class {} ioprio={} nice={} rate={} burst={}\n", + s += fmt::format("class {}{}{} ioprio={} nice={} rate={} burst={} " + "yield={} floor={}\n", c->name, + (c->protect ? " protect" : ""), + (c->critical ? " critical" : ""), qos::ioprio::to_string(c->ioprio), ((c->nice == qos::UNSET) ? "unset" : std::to_string(c->nice)), - c->rate, - c->burst); + (c->pct ? fmt::format("{}%",c->pct) : std::to_string(c->rate)), + c->burst, + c->yield, + (c->floor_pct ? fmt::format("{}%",c->floor_pct) + : std::to_string(c->floor))); } for(const auto &r : _rules) { - const char *field = ""; - switch(r.field) + s += "match"; + + for(const auto &cond : r.conditions) { - case Field::CGROUP: field = "cgroup"; break; - case Field::COMM: field = "comm"; break; - case Field::UID: field = "uid"; break; - case Field::GID: field = "gid"; break; + const char *field = ""; + switch(cond.field) + { + case Field::CGROUP: field = "cgroup"; break; + case Field::COMM: field = "comm"; break; + case Field::CMDLINE: field = "cmdline"; break; + case Field::UID: field = "uid"; break; + case Field::GID: field = "gid"; break; + case Field::PATH: field = "path"; break; + case Field::OP: field = "op"; break; + } + + s += fmt::format(" {} {} {}", + field, + ((cond.op == MatchOp::EQ) ? "=" : "~"), + cond.pattern); } - s += fmt::format("match {} {} {} -> {}\n", - field, - ((r.op == Op::EQ) ? "=" : "~"), - r.pattern, - r.cls->name); + s += fmt::format(" -> {}\n",r.cls->name); } if(_default) diff --git a/src/qos_rules.hpp b/src/qos_rules.hpp index 5afec82d..f955ee23 100644 --- a/src/qos_rules.hpp +++ b/src/qos_rules.hpp @@ -22,41 +22,73 @@ #include #include +#include #include namespace qos { - // Which attribute of the calling process a rule tests. + // Which attribute of a request a condition tests. enum class Field { - CGROUP, // contents of /proc//cgroup - COMM, // contents of /proc//comm + CGROUP, // contents of /proc//cgroup -- identifies the container + COMM, // contents of /proc//comm -- identifies the program UID, - GID + GID, + CMDLINE, // full /proc//cmdline, arguments joined by spaces + PATH, // path within the pool, e.g. /Movies/Dune (2021)/Dune.mkv + OP // "read" or "write" }; - enum class Op + enum class MatchOp { EQ, // '=' exact string / numeric equality - GLOB // '~' fnmatch(3) pattern + GLOB // '~' fnmatch(3) pattern, '/' not treated specially }; - 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 + enum class Direction + { + READ, + WRITE + }; - bool matches(const std::string &cgroup, - const std::string &comm, - const u32 uid, - const u32 gid) const; + // Everything a rule can be matched against. Assembled once per + // request, and only for the fields some rule actually uses. + struct Subject + { + const std::string *cgroup = nullptr; + const std::string *comm = nullptr; + const std::string *cmdline = nullptr; + const std::string *path = nullptr; + u32 uid = 0; + u32 gid = 0; + Direction dir = Direction::READ; }; - // An immutable, atomically published set of classes and rules. + struct Condition + { + Field field; + MatchOp op; + std::string pattern; + u32 number; // parsed form of `pattern` for UID/GID + Direction dir; // parsed form of `pattern` for OP + + bool matches(const Subject &) const; + }; + + // A rule fires when *every* one of its conditions matches, which is + // what makes "this program, but only on this folder, and only when + // writing" expressible as one rule. + struct Rule + { + std::vector conditions; + const Class *cls; // owned by the RuleSet holding this rule + + bool matches(const Subject &) const; + }; + + // An immutable, atomically published set of classes, rules and + // resource capacities. // // 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 @@ -80,31 +112,54 @@ namespace qos std::string *err); public: - const Class *classify(const std::string &cgroup, - const std::string &comm, - const u32 uid, - const u32 gid) const; + const Class *classify(const Subject &) const; const Class *find_class(const std::string_view name) const; + // Measured or declared throughput of `resource` in bytes/sec, + // falling back to the `capacity default` line. Zero when nothing + // is known, which makes percentage rates unenforceable and is + // reported as such. + u64 capacity(const std::string &resource) const; + + // Resolves a class's configured rate against the resource + // serving the request. Returns 0 for "unlimited". + u64 rate_for(const Class *, const std::string &resource) 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; } + // Which subject fields any rule actually inspects. Each one that + // is unused is a file read or a string copy skipped per request. bool needs_cgroup() const { return _needs_cgroup; } bool needs_comm() const { return _needs_comm; } + bool needs_cmdline() const { return _needs_cmdline; } + bool needs_path() const { return _needs_path; } + + // True when some class is marked `protect`, which is what turns + // the adaptive governor on. + bool has_protected() const { return _has_protected; } + + // Resolves a class's floor against the resource. + u64 floor_for(const Class *, const std::string &resource) const; 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; + std::vector> _classes; + std::vector _rules; + std::unordered_map _capacity; + u64 _capacity_default = 0; + const Class *_default = nullptr; + bool _inert = true; + bool _needs_cgroup = false; + bool _needs_comm = false; + bool _needs_cmdline = false; + bool _needs_path = false; + bool _has_protected = false; }; // Parses "10M", "1.5MiB", "512K", "1G" and bare byte counts into diff --git a/tests/tests.cpp b/tests/tests.cpp index 89ffee31..7c6e3b03 100644 --- a/tests/tests.cpp +++ b/tests/tests.cpp @@ -8,6 +8,7 @@ #include "num.hpp" #include "rapidhash/rapidhash.h" #include "rnd.hpp" +#include "qos_rules.hpp" #include "str.hpp" #include "thread_pool.hpp" @@ -3647,6 +3648,450 @@ test_rapidhash_withSeed_preserves_default_output() } } + +/* + QoS ruleset parsing. + + The parser is the part of the QoS engine a user actually touches, and + a ruleset that parses "successfully" into the wrong thing would + silently misprioritise every request. These check both that valid + input produces the expected classification and that invalid input is + rejected rather than partially applied. +*/ + +static +qos::RuleSet::Ptr +qos_parse(const char *text_) +{ + std::string err; + + return qos::RuleSet::parse(text_,&err); +} + +// Owns the strings a Subject points at. Building a Subject from +// temporaries leaves it holding dangling pointers the moment the full +// expression ends. +struct QoSSubj +{ + std::string cgroup; + std::string comm; + std::string path; + std::string cmdline; + qos::Subject s; + + QoSSubj(std::string cgroup_, + std::string comm_, + std::string path_, + const u32 uid_ = 0, + const qos::Direction dir_ = qos::Direction::READ, + std::string cmdline_ = {}) + : cgroup(std::move(cgroup_)), + comm(std::move(comm_)), + path(std::move(path_)), + cmdline(std::move(cmdline_)) + { + s.cgroup = &cgroup; + s.comm = &comm; + s.path = &path; + s.cmdline = &cmdline; + s.uid = uid_; + s.gid = 0; + s.dir = dir_; + } +}; + +static +void +test_qos_parse_size(void) +{ + u64 v; + + TEST_CHECK(qos::parse_size("1024",&v) == 0 && v == 1024); + TEST_CHECK(qos::parse_size("1K",&v) == 0 && v == 1024); + TEST_CHECK(qos::parse_size("1KiB",&v) == 0 && v == 1024); + TEST_CHECK(qos::parse_size("1MB",&v) == 0 && v == 1024 * 1024); + TEST_CHECK(qos::parse_size("1.5M",&v) == 0 && v == (1024 * 1024 * 3) / 2); + TEST_CHECK(qos::parse_size("2G",&v) == 0 && v == 2ULL * 1024 * 1024 * 1024); + + TEST_CHECK(qos::parse_size("",&v) != 0); + TEST_CHECK(qos::parse_size("M",&v) != 0); + TEST_CHECK(qos::parse_size("10Q",&v) != 0); + TEST_CHECK(qos::parse_size("abc",&v) != 0); +} + +static +void +test_qos_parse_ioprio(void) +{ + int v; + + TEST_CHECK(qos::ioprio::from_string("idle",&v) == 0); + TEST_CHECK(qos::ioprio::to_string(v) == "idle"); + + TEST_CHECK(qos::ioprio::from_string("rt:0",&v) == 0); + TEST_CHECK(qos::ioprio::to_string(v) == "rt:0"); + + TEST_CHECK(qos::ioprio::from_string("be:7",&v) == 0); + TEST_CHECK(qos::ioprio::to_string(v) == "be:7"); + + TEST_CHECK(qos::ioprio::from_string("be:8",&v) != 0); + TEST_CHECK(qos::ioprio::from_string("be:",&v) != 0); + TEST_CHECK(qos::ioprio::from_string("zz:1",&v) != 0); + TEST_CHECK(qos::ioprio::from_string("",&v) != 0); +} + +static +void +test_qos_rules_basic(void) +{ + auto rs = qos_parse("class fast ioprio=rt:0\n" + "class slow ioprio=idle nice=19 rate=8M\n" + "match comm = rsync -> slow\n" + "default fast\n"); + + TEST_ASSERT(rs != nullptr); + TEST_CHECK(!rs->inert()); + TEST_CHECK(rs->needs_comm()); + TEST_CHECK(!rs->needs_cgroup()); + TEST_CHECK(!rs->needs_path()); + + QoSSubj rsync("","rsync","/a"); + TEST_CHECK(rs->classify(rsync.s)->name == "slow"); + + QoSSubj cat("","cat","/a"); + TEST_CHECK(rs->classify(cat.s)->name == "fast"); +} + +static +void +test_qos_rules_multi_condition(void) +{ + // Every condition must hold. This is what makes "this program, but + // only on this folder" expressible as a single rule. + auto rs = qos_parse("class bulk ioprio=idle\n" + "class norm ioprio=be:4\n" + "match cgroup ~ *lxc/3111* comm ~ *Butler* -> bulk\n" + "default norm\n"); + + TEST_ASSERT(rs != nullptr); + TEST_CHECK(rs->needs_cgroup()); + TEST_CHECK(rs->needs_comm()); + + QoSSubj both("0::/lxc/3111/init","PlexButler","/x"); + TEST_CHECK(rs->classify(both.s)->name == "bulk"); + + // Right container, wrong program. + QoSSubj one("0::/lxc/3111/init","PlexTranscoder","/x"); + TEST_CHECK(rs->classify(one.s)->name == "norm"); + + // Right program, wrong container. + QoSSubj other("0::/lxc/3129/init","PlexButler","/x"); + TEST_CHECK(rs->classify(other.s)->name == "norm"); +} + +static +void +test_qos_rules_path_and_op(void) +{ + auto rs = qos_parse("class slow ioprio=idle rate=1M\n" + "class norm ioprio=be:4\n" + "match path ~ /TV/* op = write -> slow\n" + "default norm\n"); + + TEST_ASSERT(rs != nullptr); + TEST_CHECK(rs->needs_path()); + + QoSSubj w("","x","/TV/show/ep.mkv",0,qos::Direction::WRITE); + TEST_CHECK(rs->classify(w.s)->name == "slow"); + + // Same path, but a read. + QoSSubj r("","x","/TV/show/ep.mkv",0,qos::Direction::READ); + TEST_CHECK(rs->classify(r.s)->name == "norm"); + + // A write, but elsewhere. Globs are matched without FNM_PATHNAME so + // '*' spans separators. + QoSSubj e("","x","/Movies/a.mkv",0,qos::Direction::WRITE); + TEST_CHECK(rs->classify(e.s)->name == "norm"); + + QoSSubj deep("","x","/TV/a/b/c.mkv",0,qos::Direction::WRITE); + TEST_CHECK(rs->classify(deep.s)->name == "slow"); +} + +static +void +test_qos_rules_first_match_wins(void) +{ + auto rs = qos_parse("class a ioprio=rt:0\n" + "class b ioprio=idle\n" + "match comm ~ pl* -> a\n" + "match comm ~ plex -> b\n" + "default b\n"); + + TEST_ASSERT(rs != nullptr); + + QoSSubj plex("","plex","/x"); + TEST_CHECK(rs->classify(plex.s)->name == "a"); +} + +static +void +test_qos_rules_capacity_and_percent(void) +{ + auto rs = qos_parse("capacity /mnt/disk1 100M\n" + "capacity default 50M\n" + "class pct rate=10%\n" + "class abs rate=4M\n" + "match uid = 1000 -> pct\n" + "default abs\n"); + + TEST_ASSERT(rs != nullptr); + + const qos::Class *pct = rs->find_class("pct"); + const qos::Class *abs = rs->find_class("abs"); + TEST_ASSERT(pct && abs); + + TEST_CHECK(rs->capacity("/mnt/disk1") == 100 * 1024 * 1024); + // Unknown resources fall back to the declared default. + TEST_CHECK(rs->capacity("/mnt/nope") == 50 * 1024 * 1024); + + TEST_CHECK(rs->rate_for(pct,"/mnt/disk1") == (100 * 1024 * 1024) / 10); + TEST_CHECK(rs->rate_for(pct,"/mnt/nope") == (50 * 1024 * 1024) / 10); + + // An absolute rate ignores the resource entirely. + TEST_CHECK(rs->rate_for(abs,"/mnt/disk1") == 4 * 1024 * 1024); +} + +static +void +test_qos_percent_without_capacity_is_unlimited(void) +{ + // Nothing declared a capacity, so 10% of an unknown number cannot be + // enforced. Running unlimited is the only honest reading; throttling + // against a guess would be worse. + auto rs = qos_parse("class pct rate=10%\n" + "default pct\n"); + + TEST_ASSERT(rs != nullptr); + + const qos::Class *pct = rs->find_class("pct"); + TEST_ASSERT(pct != nullptr); + TEST_CHECK(pct->limited()); + TEST_CHECK(rs->rate_for(pct,"/mnt/whatever") == 0); +} + +static +void +test_qos_rules_errors(void) +{ + std::string err; + + // A rule may not name a class that does not exist yet. + TEST_CHECK(qos::RuleSet::parse("match comm = x -> nope\n",&err) == nullptr); + TEST_CHECK(!err.empty()); + + TEST_CHECK(qos::RuleSet::parse("class a ioprio=be:9\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a rate=10Q\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a nice=99\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a rate=0%\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a rate=101%\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nclass a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch bogus = x -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch comm ! x -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch comm = x a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\ndefault nope\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("frobnicate\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch uid = abc -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch uid ~ 100 -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("class a\nmatch op = sideways -> a\n",&err) == nullptr); + TEST_CHECK(qos::RuleSet::parse("capacity default\n",&err) == nullptr); +} + +static +void +test_qos_rules_comments_and_blanks(void) +{ + auto rs = qos_parse("# leading comment\n" + "\n" + "class a ioprio=rt:0 # trailing comment\n" + " \n" + "match comm = x -> a # another\n" + "default a\n"); + + TEST_ASSERT(rs != nullptr); + + QoSSubj sub("","x","/p"); + TEST_CHECK(rs->classify(sub.s)->name == "a"); +} + +static +void +test_qos_empty_ruleset_is_inert(void) +{ + // An empty or do-nothing ruleset must report itself inert so the hot + // path can skip reading /proc entirely. + auto rs = qos_parse(""); + TEST_ASSERT(rs != nullptr); + TEST_CHECK(rs->inert()); + + auto named = qos_parse("class a\ndefault a\n"); + TEST_ASSERT(named != nullptr); + TEST_CHECK(named->inert()); + + auto live = qos_parse("class a ioprio=idle\ndefault a\n"); + TEST_ASSERT(live != nullptr); + TEST_CHECK(!live->inert()); + + auto dflt = qos::RuleSet::make_default(); + TEST_ASSERT(dflt != nullptr); + TEST_CHECK(dflt->inert()); +} + +static +void +test_qos_buckets_are_per_resource(void) +{ + // A rate is a per-device allowance. Two branches must not share one + // bucket, or a class limited to 10% would be limited to 10% of one + // disk spread across all of them. + auto rs = qos_parse("class a rate=1M\ndefault a\n"); + TEST_ASSERT(rs != nullptr); + + const qos::Class *a = rs->find_class("a"); + TEST_ASSERT(a != nullptr); + + qos::Bucket *b1 = a->bucket_for("/mnt/disk1"); + qos::Bucket *b2 = a->bucket_for("/mnt/disk2"); + + TEST_CHECK(b1 != nullptr); + TEST_CHECK(b2 != nullptr); + TEST_CHECK(b1 != b2); + // Asking again returns the same bucket rather than a fresh one. + TEST_CHECK(a->bucket_for("/mnt/disk1") == b1); +} + + + +static +void +test_qos_rules_cmdline(void) +{ + // The case this field exists for: Plex runs credits detection under + // the *same* `Plex Transcoder` binary as real playback, reading the + // same media file. Neither comm nor path can separate them -- only + // the command line, where a detection job writes its output into + // .../Transcode/Detection/ instead of feeding a session. + auto rs = qos_parse("class playback protect\n" + "class analysis yield=100 floor=5% ioprio=idle\n" + "match comm = 'Plex Transcoder' cmdline ~ */Transcode/Detection/* -> analysis\n" + "match comm = 'Plex Transcoder' -> playback\n" + "default playback\n"); + + TEST_ASSERT(rs != nullptr); + TEST_CHECK(rs->needs_cmdline()); + + QoSSubj detect("","Plex Transcoder","/TV/ep.mkv",0,qos::Direction::READ, + "/usr/lib/plexmediaserver/Plex Transcoder -i /storage/TV/ep.mkv " + "-f flac /transcode/Transcode/Detection/abc123"); + TEST_CHECK(rs->classify(detect.s)->name == "analysis"); + + // Same binary, same file, but feeding a session. + QoSSubj play("","Plex Transcoder","/TV/ep.mkv",0,qos::Direction::READ, + "/usr/lib/plexmediaserver/Plex Transcoder -i /storage/TV/ep.mkv " + "-f hls /transcode/Transcode/Sessions/xyz789"); + TEST_CHECK(rs->classify(play.s)->name == "playback"); +} + +static +void +test_qos_cmdline_not_read_when_unused(void) +{ + // Reading /proc//cmdline is only worth it when some rule looks + // at it. + auto rs = qos_parse("class a ioprio=idle\n" + "match comm = x -> a\n" + "default a\n"); + + TEST_ASSERT(rs != nullptr); + TEST_CHECK(!rs->needs_cmdline()); +} + +static +void +test_qos_critical_is_never_throttled(void) +{ + // A class playback synchronously waits on -- Plex's EasyAudioEncoder + // is the real-world case -- must never be delayed, even if the + // ruleset gives it a rate. Throttling it throttles the stream + // blocked behind it. + auto rs = qos_parse("class eae critical rate=1M\n" + "class bulk rate=1M\n" + "match comm = EasyAudioEncode -> eae\n" + "default bulk\n"); + + TEST_ASSERT(rs != nullptr); + + const qos::Class *eae = rs->find_class("eae"); + const qos::Class *bulk = rs->find_class("bulk"); + TEST_ASSERT(eae && bulk); + + TEST_CHECK(eae->critical); + TEST_CHECK(eae->immune()); + TEST_CHECK(!bulk->immune()); + + QoSSubj s("","EasyAudioEncode","/x"); + TEST_CHECK(rs->classify(s.s)->name == "eae"); +} + +static +void +test_qos_protect_implies_immune(void) +{ + auto rs = qos_parse("class p protect\nclass b yield=100\ndefault b\n"); + + TEST_ASSERT(rs != nullptr); + + TEST_CHECK(rs->find_class("p")->immune()); + TEST_CHECK(!rs->find_class("b")->immune()); + TEST_CHECK(rs->has_protected()); +} + + + +static +void +test_qos_quoted_patterns(void) +{ + // Without quoting, none of the process names that actually matter + // here can be written down at all. + auto rs = qos_parse("class a ioprio=idle\n" + "class b ioprio=be:4\n" + "match comm = \"Plex Media Scanner\" -> a\n" + "default b\n"); + + TEST_ASSERT(rs != nullptr); + + QoSSubj scanner("","Plex Media Scanner","/x"); + TEST_CHECK(rs->classify(scanner.s)->name == "a"); + + QoSSubj other("","Plex","/x"); + TEST_CHECK(rs->classify(other.s)->name == "b"); + + // Single quotes work the same way, and a glob may be quoted too. + auto rs2 = qos_parse("class a ioprio=idle\n" + "class b ioprio=be:4\n" + "match path ~ '/TV/The Bear/*' -> a\n" + "default b\n"); + TEST_ASSERT(rs2 != nullptr); + + QoSSubj bear("","x","/TV/The Bear/S01E01.mkv"); + TEST_CHECK(rs2->classify(bear.s)->name == "a"); + + QoSSubj notbear("","x","/TV/Other/S01E01.mkv"); + TEST_CHECK(rs2->classify(notbear.s)->name == "b"); +} + + TEST_LIST = { {"nop",test_nop}, @@ -3828,5 +4273,22 @@ TEST_LIST = {"tp_heavy_repeated_construct_destroy",test_tp_heavy_repeated_construct_destroy}, {"tp_heavy_enqueue_task_mixed_outcomes",test_tp_heavy_enqueue_task_mixed_outcomes}, {"tp_heavy_add_remove_churn_under_enqueue",test_tp_heavy_add_remove_churn_under_enqueue}, + {"qos_parse_size",test_qos_parse_size}, + {"qos_parse_ioprio",test_qos_parse_ioprio}, + {"qos_rules_basic",test_qos_rules_basic}, + {"qos_rules_multi_condition",test_qos_rules_multi_condition}, + {"qos_rules_path_and_op",test_qos_rules_path_and_op}, + {"qos_rules_first_match_wins",test_qos_rules_first_match_wins}, + {"qos_rules_capacity_and_percent",test_qos_rules_capacity_and_percent}, + {"qos_percent_without_capacity_is_unlimited",test_qos_percent_without_capacity_is_unlimited}, + {"qos_rules_errors",test_qos_rules_errors}, + {"qos_rules_comments_and_blanks",test_qos_rules_comments_and_blanks}, + {"qos_empty_ruleset_is_inert",test_qos_empty_ruleset_is_inert}, + {"qos_buckets_are_per_resource",test_qos_buckets_are_per_resource}, + {"qos_quoted_patterns",test_qos_quoted_patterns}, + {"qos_rules_cmdline",test_qos_rules_cmdline}, + {"qos_cmdline_not_read_when_unused",test_qos_cmdline_not_read_when_unused}, + {"qos_critical_is_never_throttled",test_qos_critical_is_never_throttled}, + {"qos_protect_implies_immune",test_qos_protect_implies_immune}, {NULL,NULL} }; diff --git a/tools/mergerfs.qos-bench b/tools/mergerfs.qos-bench new file mode 100755 index 00000000..c170e667 --- /dev/null +++ b/tools/mergerfs.qos-bench @@ -0,0 +1,314 @@ +#!/usr/bin/env python3 +""" +Measure the throughput of each branch in a mergerfs pool and emit the +`capacity` lines a QoS ruleset needs to turn percentage rates into +numbers. + +A percentage rate ("give the scanner 10% of this disk") is meaningless +until something knows what 100% is. This measures it. + +Defaults are chosen to be safe to run against a pool that is in use: + + * Reads only. No file is created or modified unless --allow-write is + given, and nothing is ever written into the pool itself. + * O_DIRECT, so the measurement neither reads stale numbers out of the + page cache nor evicts anything the running services still want. + The system-wide cache is never dropped. + * Idle I/O priority and nice 19, so a benchmark that collides with + playback yields rather than competes. Note this also means the + number produced is a floor when the pool is busy -- for a true + ceiling, measure when it is quiet. + * Short samples (3s per branch by default). +""" + +import argparse +import ctypes +import errno +import mmap +import os +import random +import sys +import time + +BLOCK = 4 * 1024 * 1024 # O_DIRECT reads must be sector aligned +MIN_PROBE_SIZE = 256 * 1024 * 1024 +MAX_SCAN_ENTRIES = 20000 + +IOPRIO_WHO_PROCESS = 1 +IOPRIO_CLASS_IDLE = 3 +IOPRIO_CLASS_SHIFT = 13 + + +def be_polite(): + """Drop to idle I/O priority and nice 19.""" + try: + os.nice(19) + except OSError: + pass + try: + libc = ctypes.CDLL("libc.so.6", use_errno=True) + libc.syscall(251, IOPRIO_WHO_PROCESS, 0, + IOPRIO_CLASS_IDLE << IOPRIO_CLASS_SHIFT) + except Exception: + pass + + +def human(n): + for unit in ("B", "K", "M", "G", "T"): + if n < 1024 or unit == "T": + return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}" + n /= 1024.0 + + +def branches_of(mountpoint): + """Read the branch list out of a live mergerfs mount.""" + try: + raw = os.getxattr(os.path.join(mountpoint, ".mergerfs"), + "user.mergerfs.branches") + except OSError as e: + sys.exit(f"{mountpoint}: cannot read branch list: {e}") + + out = [] + for spec in raw.decode().split(":"): + # Each entry is =; mode is irrelevant here. + out.append(spec.split("=", 1)[0]) + return out + + +def find_probe_file(branch, min_size): + """ + Find one file big enough to read a sample from. + + Walks breadth-first with a hard cap on entries examined; on a 14TB + disk of media the first large file turns up almost immediately, and + the cap keeps a pathological layout from turning this into a full + tree walk. + """ + seen = 0 + queue = [branch] + + while queue: + d = queue.pop(0) + try: + with os.scandir(d) as it: + for entry in it: + seen += 1 + if seen > MAX_SCAN_ENTRIES: + return None + try: + if entry.is_symlink(): + continue + if entry.is_dir(): + queue.append(entry.path) + elif entry.is_file(): + if entry.stat().st_size >= min_size: + return entry.path + except OSError: + continue + except OSError: + continue + + return None + + +def measure_read(path, duration, block=BLOCK): + """ + Sequential O_DIRECT read from a random aligned offset. + + Returns bytes/sec, or None if the file could not be read directly. + """ + try: + fd = os.open(path, os.O_RDONLY | os.O_DIRECT) + except OSError as e: + if e.errno in (errno.EINVAL, errno.ENOTSUP, errno.EOPNOTSUPP): + return None + raise + + try: + size = os.fstat(fd).st_size + if size < block * 2: + return None + + # Start somewhere other than the front of the file so repeated + # runs do not all measure the same tracks. + span = size - block + start = random.randrange(0, max(1, span // block)) * block + + buf = mmap.mmap(-1, block) # mmap is page aligned + total = 0 + offset = start + deadline = time.monotonic() + duration + t0 = time.monotonic() + + while time.monotonic() < deadline: + if offset + block > size: + offset = 0 + n = os.preadv(fd, [buf], offset) + if n <= 0: + break + total += n + offset += block + + elapsed = time.monotonic() - t0 + buf.close() + + if elapsed <= 0 or total == 0: + return None + + return total / elapsed + finally: + os.close(fd) + + +def measure_write(branch, duration, block=BLOCK): + """Sequential O_DIRECT write to a temp file on the branch.""" + path = os.path.join(branch, f".mergerfs-qos-bench.{os.getpid()}") + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_DIRECT, 0o600) + except OSError as e: + print(f" write: cannot create probe file: {e}", file=sys.stderr) + return None + + try: + buf = mmap.mmap(-1, block) + buf.write(os.urandom(65536) * (block // 65536)) + total = 0 + deadline = time.monotonic() + duration + t0 = time.monotonic() + + while time.monotonic() < deadline: + n = os.pwritev(fd, [buf], total) + if n <= 0: + break + total += n + + elapsed = time.monotonic() - t0 + buf.close() + + if elapsed <= 0 or total == 0: + return None + return total / elapsed + finally: + os.close(fd) + try: + os.unlink(path) + except OSError: + pass + + +def main(): + ap = argparse.ArgumentParser( + description="Measure per-branch throughput for a mergerfs QoS ruleset.") + ap.add_argument("target", + help="a mergerfs mountpoint, or one or more branch paths") + ap.add_argument("extra", nargs="*", + help="additional branch paths") + ap.add_argument("-d", "--duration", type=float, default=3.0, + help="seconds to sample each branch (default: 3)") + ap.add_argument("-o", "--output", + help="write the capacity block to this file instead of stdout") + ap.add_argument("--write", action="store_true", + help="also measure write throughput (creates a temp file " + "on each branch)") + ap.add_argument("--allow-write", action="store_true", + help="permit creating a probe file when no existing file " + "is large enough to read from") + ap.add_argument("--min-size", type=int, default=MIN_PROBE_SIZE, + help="smallest file usable as a read probe, in bytes") + ap.add_argument("--full-speed", action="store_true", + help="do not lower this process's priority; measures the " + "true ceiling but competes with live traffic") + args = ap.parse_args() + + if not args.full_speed: + be_polite() + + # A single argument that is a mergerfs mount means "discover the + # branches yourself". + if not args.extra and os.path.exists(os.path.join(args.target, ".mergerfs")): + branches = branches_of(args.target) + print(f"# discovered {len(branches)} branches via {args.target}", + file=sys.stderr) + else: + branches = [args.target] + args.extra + + results = {} + + for branch in branches: + if not os.path.isdir(branch): + print(f"{branch}: not a directory, skipping", file=sys.stderr) + continue + + print(f"{branch}:", file=sys.stderr, flush=True) + + probe = find_probe_file(branch, args.min_size) + created = None + + if probe is None: + if not args.allow_write: + print(f" no file >= {human(args.min_size)} found; " + f"skipping (use --allow-write to create a probe file)", + file=sys.stderr) + continue + created = os.path.join(branch, f".mergerfs-qos-probe.{os.getpid()}") + try: + with open(created, "wb") as f: + chunk = os.urandom(1 << 20) + for _ in range(args.min_size >> 20): + f.write(chunk) + f.flush() + os.fsync(f.fileno()) + probe = created + except OSError as e: + print(f" cannot create probe file: {e}", file=sys.stderr) + continue + + try: + rd = measure_read(probe, args.duration) + if rd is None: + print(" read: O_DIRECT unavailable or file too small", + file=sys.stderr) + else: + print(f" read: {human(rd)}/s (probe: {probe})", + file=sys.stderr) + results[branch] = rd + + if args.write: + wr = measure_write(branch, args.duration) + if wr is not None: + print(f" write: {human(wr)}/s", file=sys.stderr) + finally: + if created: + try: + os.unlink(created) + except OSError: + pass + + if not results: + sys.exit("no branches could be measured") + + lines = ["# generated by mergerfs.qos-bench on " + + time.strftime("%Y-%m-%d %H:%M:%S"), + "# read throughput, O_DIRECT, " + f"{args.duration:g}s sample per branch"] + if not args.full_speed: + lines.append("# measured at idle priority: treat as a floor if the " + "pool was busy") + for branch, rate in results.items(): + lines.append(f"capacity {branch} {int(rate)}") + + total = sum(results.values()) + lines.append(f"capacity default {int(total / len(results))}") + + block = "\n".join(lines) + "\n" + + if args.output: + with open(args.output, "w") as f: + f.write(block) + print(f"\nwrote {args.output}", file=sys.stderr) + else: + sys.stdout.write(block) + + +if __name__ == "__main__": + main() diff --git a/tools/mergerfs.qos-playback-test b/tools/mergerfs.qos-playback-test new file mode 100755 index 00000000..fbaf4bee --- /dev/null +++ b/tools/mergerfs.qos-playback-test @@ -0,0 +1,324 @@ +#!/usr/bin/env python3 +""" +Prove that playback survives competing load -- or show exactly where it +does not. + +A rate limit that looks right in a config file is worth nothing if a +film still stutters. This models a player: it reads a real file out of +the pool at a fixed bitrate through a jitter buffer, and counts the +moments the buffer ran dry, which is what a viewer experiences as a +stutter. It can start and stop competing bulk load partway through, so +one run shows three things at once: + + 1. baseline -- what the buffer does with the pool to itself + 2. contended -- whether bulk load empties it, and how far pressure + had to rise to stop that + 3. recovery -- whether the yielding classes climb back to full speed + once playback ends + +Both the player and the load generator rename themselves, so a ruleset +can classify them by name: + + class playback protect ioprio=rt:0 + class bulk yield=100 floor=5% ioprio=idle + match comm = qos-playback -> playback + match comm = qos-bulk -> bulk +""" + +import argparse +import ctypes +import os +import statistics +import sys +import threading +import time + +CHUNK = 1 << 20 +PR_SET_NAME = 15 + + +def set_comm(name): + """Rename this thread so mergerfs QoS rules can match on comm.""" + try: + libc = ctypes.CDLL("libc.so.6", use_errno=True) + libc.prctl(PR_SET_NAME, ctypes.c_char_p(name.encode()[:15]), 0, 0, 0) + except Exception: + pass + + +def human(n): + for unit in ("B", "K", "M", "G"): + if n < 1024 or unit == "G": + return f"{n:.1f}{unit}" + n /= 1024.0 + + +def parse_size(s): + s = s.strip() + mult = 1 + if s and s[-1] in "kKmMgG": + mult = {"k": 1024, "m": 1024 ** 2, "g": 1024 ** 3}[s[-1].lower()] + s = s[:-1] + return int(float(s) * mult) + + +class Player: + """ + Reads `path` at `bitrate`, keeping a jitter buffer topped up. + + The buffer is the whole point: a player does not care how fast a + read was, only whether the next frame was there in time. Every + moment the buffer hits zero is a stutter. + """ + + def __init__(self, path, bitrate, buffer_secs): + self.path = path + self.bitrate = bitrate + self.target = int(bitrate * buffer_secs) + self.underruns = 0 + self.min_buffer = self.target + self.latencies = [] + self.bytes_read = 0 + self.stop = threading.Event() + self.samples = [] # (t, buffer_bytes) + + def run(self): + set_comm("qos-playback") + + fd = os.open(self.path, os.O_RDONLY) + size = os.fstat(fd).st_size + offset = 0 + + start = time.monotonic() + # Prefilled, as a player would be after it starts. + fetched = self.target + in_underrun = False + + try: + while not self.stop.is_set(): + now = time.monotonic() + elapsed = now - start + consumed = elapsed * self.bitrate + buffer = fetched - consumed + + self.min_buffer = min(self.min_buffer, buffer) + self.samples.append((elapsed, buffer)) + + if buffer <= 0: + # The buffer is empty: the viewer is looking at a + # spinner. Count it once per drain, not once per + # loop iteration. + if not in_underrun: + self.underruns += 1 + in_underrun = True + else: + in_underrun = False + + if buffer >= self.target: + # Buffer is full; a real player would idle here + # rather than race ahead. + time.sleep(0.005) + continue + + if offset + CHUNK > size: + offset = 0 + + t0 = time.monotonic() + data = os.pread(fd, CHUNK, offset) + dt = time.monotonic() - t0 + + if not data: + offset = 0 + continue + + self.latencies.append(dt) + fetched += len(data) + self.bytes_read += len(data) + offset += len(data) + finally: + os.close(fd) + + +class Bulk: + """A greedy reader, standing in for a download or a library scan.""" + + def __init__(self, path): + self.path = path + self.bytes_read = 0 + self.stop = threading.Event() + + def run(self): + set_comm("qos-bulk") + + fd = os.open(self.path, os.O_RDONLY) + size = os.fstat(fd).st_size + offset = 0 + try: + while not self.stop.is_set(): + if offset + CHUNK > size: + offset = 0 + data = os.pread(fd, CHUNK, offset) + if not data: + offset = 0 + continue + self.bytes_read += len(data) + offset += len(data) + finally: + os.close(fd) + + +def read_pressure(mountpoint): + """Pull the governor's view out of the live mount.""" + if not mountpoint: + return {} + try: + raw = os.getxattr(os.path.join(mountpoint, ".mergerfs"), + "user.mergerfs.qos.stats").decode() + except OSError: + return {} + + out = {} + for line in raw.splitlines(): + if line.startswith("resource "): + name = line.split()[1].rstrip(":") + fields = dict(f.split("=", 1) for f in line.split() if "=" in f) + out[name] = fields + return out + + +def main(): + ap = argparse.ArgumentParser( + description="Measure whether playback survives competing pool load.") + ap.add_argument("file", help="a media file inside the pool to play") + ap.add_argument("--bulk-file", action="append", default=[], + help="file for a competing bulk reader (repeatable)") + ap.add_argument("-b", "--bitrate", default="25M", + help="playback bitrate, e.g. 25M (default: 25M)") + ap.add_argument("--buffer", type=float, default=5.0, + help="jitter buffer in seconds (default: 5)") + ap.add_argument("-d", "--duration", type=float, default=60.0, + help="total run time in seconds (default: 60)") + ap.add_argument("--bulk-start", type=float, default=10.0, + help="when to start the bulk load (default: 10s)") + ap.add_argument("--bulk-stop", type=float, default=None, + help="when to stop it (default: 10s before the end)") + ap.add_argument("-m", "--mount", + help="mergerfs mountpoint, to report governor pressure") + args = ap.parse_args() + + bitrate = parse_size(args.bitrate) + bulk_stop = (args.bulk_stop if args.bulk_stop is not None + else max(args.bulk_start, args.duration - 10.0)) + + if not args.bulk_file: + args.bulk_file = [args.file] + + # Check up front: a missing file would otherwise surface as a + # traceback from a worker thread while the run carried on + # reporting zero bulk load, which reads like a passing test. + for path in [args.file] + args.bulk_file: + if not os.path.isfile(path): + sys.exit(f"{path}: not a file") + + player = Player(args.file, bitrate, args.buffer) + pthread = threading.Thread(target=player.run, daemon=True) + + print(f"playing {args.file} at {human(bitrate)}/s " + f"with a {args.buffer:g}s buffer for {args.duration:g}s") + print(f"bulk load: {len(args.bulk_file)} reader(s), " + f"{args.bulk_start:g}s -> {bulk_stop:g}s\n") + print(f"{'t':>6} {'phase':<10} {'buffer':>9} {'play':>10} {'bulk':>10} " + f"{'pressure':>9} {'under':>6}") + + pthread.start() + + bulks = [] + bthreads = [] + started = False + stopped = False + + t0 = time.monotonic() + last_play = 0 + last_bulk = 0 + last_t = 0.0 + + while True: + now = time.monotonic() - t0 + if now >= args.duration: + break + + if (not started) and (now >= args.bulk_start): + for bf in args.bulk_file: + b = Bulk(bf) + t = threading.Thread(target=b.run, daemon=True) + bulks.append(b) + bthreads.append(t) + t.start() + started = True + + if started and (not stopped) and (now >= bulk_stop): + for b in bulks: + b.stop.set() + stopped = True + + time.sleep(1.0) + + now = time.monotonic() - t0 + dt = max(1e-9, now - last_t) + + play_total = player.bytes_read + bulk_total = sum(b.bytes_read for b in bulks) + + play_rate = (play_total - last_play) / dt + bulk_rate = (bulk_total - last_bulk) / dt + + last_play, last_bulk, last_t = play_total, bulk_total, now + + buffer = player.samples[-1][1] if player.samples else 0 + + press = read_pressure(args.mount) + # Report the most-pressured resource; that is the one doing the + # protecting. + pval = "-" + if press: + vals = [] + for f in press.values(): + try: + vals.append(float(f.get("pressure", 0))) + except ValueError: + pass + if vals: + pval = f"{max(vals):.2f}" + + phase = ("baseline" if not started + else ("recovery" if stopped else "contended")) + + print(f"{now:6.0f} {phase:<10} {human(max(0,buffer)):>9} " + f"{human(play_rate)+'/s':>10} {human(bulk_rate)+'/s':>10} " + f"{pval:>9} {player.underruns:>6}") + + player.stop.set() + for b in bulks: + b.stop.set() + pthread.join(timeout=5) + + print() + lat = sorted(player.latencies) + if lat: + def pct(p): + return lat[min(len(lat) - 1, int(len(lat) * p))] * 1000 + + print(f"read latency: median {statistics.median(lat)*1000:.1f}ms " + f"p95 {pct(0.95):.1f}ms p99 {pct(0.99):.1f}ms " + f"max {lat[-1]*1000:.1f}ms") + + print(f"buffer low-water: {human(max(0,player.min_buffer))} " + f"of {human(player.target)}") + print(f"underruns (stutters): {player.underruns}") + + # A stutter is the only failure that matters here. + sys.exit(1 if player.underruns else 0) + + +if __name__ == "__main__": + main()