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()