qos: match on cmdline, and never throttle critical dependencies

Two gaps found while using this against a real media server.

Some programs do very different jobs under one name. Plex runs credits
detection with the same 'Plex Transcoder' binary that serves playback,
reading the same file, so neither comm nor path separates them. Adds a
cmdline match field, and quoting so a pattern can contain spaces --
without which 'Plex Transcoder' cannot be written down at all.

Adds a 'critical' class flag for services playback synchronously waits
on. Plex's EasyAudioEncoder is the case: a transcode of EAC3 audio
writes to EAE and blocks. Such a helper looks like background work, so
a governor protecting playback will throttle or suspend it and hang the
stream waiting behind it. A critical class is never delayed.

Also fixes qos.max-sleepers auto-sizing, which read the requested
thread count rather than the resolved one and so capped at 1.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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 <name> [protect] [ioprio=…] [nice=…] [rate=…] [burst=…] [yield=…] [floor=…]
+
+match <field> <op> <pattern> [<field> <op> <pattern> …] -> <class>
+
+default <class>
+```
+
+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/<pid>/cgroup` -- identifies the container |
+| `comm` | `/proc/<pid>/comm` -- identifies the program |
+| `cmdline` | `/proc/<pid>/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/<pid>/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.

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

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;

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<int>    read_thread_count;

src/config_qos.cpp

@@ -22,7 +22,10 @@
 #include "qos.hpp"
 #include "syslog.hpp"
 
+#include "fmt/core.h"
+
 #include <errno.h>
+#include <stdlib.h>
 
 
 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<u64>(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;
+}

src/config_qos.hpp

@@ -71,6 +71,22 @@ public:
   int from_string(const std::string_view) final;
 };
 
+// qos.distress-ms=<int>
+class QoSDistressMS : public ToFromString
+{
+public:
+  std::string to_string(void) const final;
+  int from_string(const std::string_view) final;
+};
+
+// qos.distress-factor=<float>
+class QoSDistressFactor : public ToFromString
+{
+public:
+  std::string to_string(void) const final;
+  int from_string(const std::string_view) final;
+};
+
 // qos.max-sleep-ms=<int>
 class QoSMaxSleepMS : public ToFromString
 {

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)
+  // Classified after the file is known so path rules can see which
-    return ::_read_direct_io(fi->fd,buf_,size_,offset_);
+  // 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

src/fuse_write.cpp

@@ -129,18 +129,11 @@ _move_and_pwriten(char const    *buf_,
 
 static
 int
-_write(const fuse_req_ctx_t   *ctx_,
+_write(FileInfo     *fi,
-       const fuse_file_info_t *ffi_,
+       const char   *buf_,
-       const char             *buf_,
+       const size_t  count_,
-       const size_t            count_,
+       const off_t   offset_)
-       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

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<char,4096> buf;
+  fmt::format_to_n_result<char*> 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();
+}

src/procfs.hpp

@@ -31,4 +31,10 @@ namespace procfs
   // Raw contents of /proc/<pid>/cgroup with the trailing newline
   // removed. Empty when the process is gone or unreadable.
   std::string get_cgroup(const int pid);
+
+  // /proc/<pid>/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);
 }

src/qos.cpp

@@ -18,6 +18,7 @@
 
 #include "qos.hpp"
 
+#include "config.hpp"
 #include "ioprio.hpp"
 #include "procfs.hpp"
 
@@ -27,12 +28,19 @@
 #include <sys/resource.h>
 #include <time.h>
 
+#include <algorithm>
+#include <cstdlib>
 #include <fstream>
+#include <memory>
 #include <mutex>
 #include <sstream>
+#include <thread>
+#include <unordered_map>
 
 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<int> 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<int> max_sleepers{-1};
   std::atomic<u64> max_sleep_ns{50ULL * 1000 * 1000};
+
+  std::atomic<u64>    distress_floor_ns{50ULL * 1000 * 1000};
+  std::atomic<double> 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<std::string,std::unique_ptr<qos::Governor>> g_governors;
+
+  qos::Governor *
+  _governor_for(const std::string &resource_)
+  {
+    std::lock_guard<std::mutex> 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<qos::Governor>());
+
+    return it->second.get();
+  }
+
   std::mutex          g_mutex;
   std::atomic<u64>    g_generation{1};
   std::atomic<int>    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<int>(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 *
+  // The cache is keyed by the whole subject, not just the pid: with
-  _lookup(const u32 pid_,
+  // path rules in play the same process reading two different folders
-          const u64 now_)
+  // 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))
+        if(e.valid && (e.pid == pid_) && (e.key == key_) && (now_ < e.expire))
-          return e.cls;
+          {
+            *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<std::string>{}(*path_) << 1);
+
+    return key;
+  }
 }
 
 void
@@ -215,56 +326,259 @@ qos::load_file(const std::string &path_,
   return 0;
 }
 
-void
+u64
-qos::throttle(const Class *cls_,
+qos::timing_start(const Apply &apply_)
-              const u64    bytes_)
 {
-  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);
+  const u64 now = ::_now_ns();
-  cls_->bytes.fetch_add(bytes_,std::memory_order_relaxed);
+  const u64 latency = ((now > started_) ? (now - started_) : 0);
 
-  if(cls_->rate == 0)
+  qos::Governor *g = ::_governor_for(resource_);
+
+  std::lock_guard<std::mutex> 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<u64>(static_cast<u64>(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<double>(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<std::mutex> 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<std::mutex> 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<double>(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<u64>(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<std::mutex> lk(cls_->bucket_mutex);
+    std::lock_guard<std::mutex> lk(bucket->mutex);
 
     const u64 now = ::_now_ns();
 
-    if(cls_->refilled == 0)
+    if(!bucket->primed)
-      cls_->refilled = now;
+      {
+        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<unsigned __int128>(cls_->rate)) / NS_PER_SEC);
+          ((elapsed * static_cast<unsigned __int128>(rate)) / NS_PER_SEC);
 
-        if(added >= cls_->burst)
+        if(added >= burst)
-          cls_->tokens = cls_->burst;
+          bucket->tokens = burst;
         else
-          cls_->tokens = std::min<u64>(cls_->burst,
+          bucket->tokens = std::min<u64>(burst,
-                                       cls_->tokens + static_cast<u64>(added));
+                                         bucket->tokens + static_cast<u64>(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<u64>((static_cast<unsigned __int128>(deficit) *
-                                          NS_PER_SEC) / cls_->rate);
+                                          NS_PER_SEC) / rate);
 
-    cls_->refilled += need_ns;
+    bucket->refilled += need_ns;
-    if(cls_->refilled > (now + MAX_DEBT_NS))
+    if(bucket->refilled > (now + MAX_DEBT_NS))
-      cls_->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.fetch_add(1,std::memory_order_relaxed);
-  cls_->throttled_ns.fetch_add(sleep_ns,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);
+  const u64 now = ::_now_ns();
-  if(cls == nullptr)
+  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);
+      Subject subject;
-      ::_remember(ctx_->pid,now,cls);
+      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<std::mutex> lk(g_gov_mutex);
+    const u64 now = ::_now_ns();
+
+    for(const auto &[resource,g] : g_governors)
+      {
+        std::lock_guard<std::mutex> 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<double>(g->latency_ewma) / 1000000.0),
+                           (static_cast<double>(g->latency_base) / 1000000.0),
+                           g->distress_events);
+      }
+  }
+
   for(const auto &c : rs->classes())
     {
       out += fmt::format("{}: requests={} bytes={} throttled={} "

src/qos.hpp

@@ -88,16 +88,60 @@ namespace qos
   extern std::atomic<int> max_sleepers;
   extern std::atomic<u64> 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<u64>    distress_floor_ns;
+  extern std::atomic<double> 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,
+  // Feedback loop.
-  // sleeps for as long as the bounds above permit. Called with the
+  //
-  // size the caller asked for, before the I/O is issued.
+  // A protected class's service time is the control signal. When reads
-  void throttle(const Class *, const u64 bytes);
+  // 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;
   };
 }

src/qos_class.cpp

@@ -0,0 +1,37 @@
+/*
+  ISC License
+
+  Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
+
+  Permission to use, copy, modify, and/or distribute this software for any
+  purpose with or without fee is hereby granted, provided that the above
+  copyright notice and this permission notice appear in all copies.
+
+  THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+  WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+  MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+  ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+  WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+  ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+  OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include "qos_class.hpp"
+
+
+qos::Bucket *
+qos::Class::bucket_for(const std::string &resource_) const
+{
+  std::lock_guard<std::mutex> 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<Bucket>());
+
+  return it->second.get();
+}

src/qos_class.hpp

@@ -23,6 +23,8 @@
 #include <atomic>
 #include <mutex>
 #include <string>
+#include <memory>
+#include <unordered_map>
 
 
 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
+  // The descriptive members are immutable once a ruleset is
-  // is published; the descriptive members are immutable for the life
+  // published, which is what lets the classifier hand out bare
-  // of the object, which is what lets the classifier hand out bare
+  // pointers to it. Everything mutable is either atomic or guarded.
-  // pointers to it.
   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
+    // A rate is either absolute (`rate` bytes/sec) or a share of the
-    // second of `rate` when a rate is set.
+    // 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
+    // Buckets are created on first use and keyed by resource name
-    // and `refilled` is a CLOCK_MONOTONIC nanosecond stamp. All three
+    // (a branch path, or "pool" when the branch is unknown). Seven
-    // are only touched when `rate` is non-zero, so classes without a
+    // branches means seven entries, so a plain map under a mutex is
-    // rate never take the lock.
+    // cheaper than anything cleverer.
-    mutable std::mutex       bucket_mutex;
+    mutable std::mutex                           buckets_mutex;
-    mutable u64              tokens{0};
+    mutable std::unordered_map<std::string,
-    mutable u64              refilled{0};
+                               std::unique_ptr<Bucket>> buckets;
+
+    Bucket *bucket_for(const std::string &resource) const;
 
   public:
     // Counters, reported via the qos.stats config key.

src/qos_rules.cpp

@@ -28,10 +28,13 @@
 #include <sstream>
 
 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_,
+qos::Condition::matches(const Subject &s_) const
-              const std::string &comm_,
-              const u32          uid_,
-              const u32          gid_) 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)
+  if(op == MatchOp::EQ)
-    return (subject == pattern);
+    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_,
+RuleSet::classify(const Subject &s_) const
-                  const std::string &comm_,
-                  const u32          uid_,
-                  const u32          gid_) 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<std::string>
   tokenize(const std::string &line_)
   {
     std::vector<std::string> 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<unsigned char>(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))
+                  if(!val.empty() && (val.back() == '%'))
-                    return fail(fmt::format("invalid rate '{}'",val));
+                    {
+                      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<u32>(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<u32>(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<u32>(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.
+          // without letting an idle class bank unlimited credit. For
-          if(cls->rate && (cls->burst == 0))
+          // a percentage rate the depth can only be resolved once the
-            cls->burst = cls->rate;
+          // serving resource is known, so it is left at zero here.
-          if(cls->burst < cls->rate)
+          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 <branch-path|default> <rate>");
+
+          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 <field> <=|~> <pattern> -> <class>");
-          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))
+          // Conditions are (field, operator, pattern) triples running
-            return fail(fmt::format("unknown field '{}'",tokens[1]));
+          // up to the arrow. Every one must match for the rule to
-
+          // fire.
-          if(tokens[2] == "=")
+          while((i < tokens.size()) && (tokens[i] != "->"))
-            rule.op = Op::EQ;
-          else if(tokens[2] == "~")
-            rule.op = Op::GLOB;
-          else
-            return fail(fmt::format("unknown operator '{}'",tokens[2]));
-
-          if((rule.field == Field::UID) || (rule.field == Field::GID))
             {
-              if(rule.op != Op::EQ)
+              if((i + 2) >= tokens.size())
-                return fail("uid/gid only support the '=' operator");
+                return fail("expected: match <field> <=|~> <pattern> ... -> <class>");
 
-              char *end = nullptr;
+              Condition cond{};
-              const unsigned long n = ::strtoul(tokens[3].c_str(),&end,10);
+
-              if((end == tokens[3].c_str()) || (*end != '\0'))
+              if(::parse_field(tokens[i],&cond.field))
-                return fail(fmt::format("invalid uid/gid '{}'",tokens[3]));
+                return fail(fmt::format("unknown field '{}'",tokens[i]));
-              rule.number = static_cast<u32>(n);
+
+              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<u32>(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]));
+                                    "defined before use)",tokens[i+1]));
-
-          if(rule.field == Field::CGROUP)
-            rs->_needs_cgroup = true;
-          if(rule.field == Field::COMM)
-            rs->_needs_comm = true;
 
           rs->_rules.push_back(std::move(rule));
         }
@@ -407,8 +648,13 @@ RuleSet::parse(const std::string_view text_,
         }
     }
 
-  if(rs->_rules.empty() && rs->_default->ioprio == qos::UNSET &&
+  if(rs->_rules.empty() &&
-     rs->_default->nice == qos::UNSET && rs->_default->rate == 0)
+     (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->pct ? fmt::format("{}%",c->pct) : std::to_string(c->rate)),
-                       c->burst);
+                       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 = "";
+      s += "match";
-      switch(r.field)
+
+      for(const auto &cond : r.conditions)
         {
-        case Field::CGROUP: field = "cgroup"; break;
+          const char *field = "";
-        case Field::COMM:   field = "comm";   break;
+          switch(cond.field)
-        case Field::UID:    field = "uid";    break;
+            {
-        case Field::GID:    field = "gid";    break;
+            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",
+      s += fmt::format(" -> {}\n",r.cls->name);
-                       field,
-                       ((r.op == Op::EQ) ? "=" : "~"),
-                       r.pattern,
-                       r.cls->name);
     }
 
   if(_default)

src/qos_rules.hpp

@@ -22,41 +22,73 @@
 
 #include <memory>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 
 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/<pid>/cgroup
+     CGROUP,   // contents of /proc/<pid>/cgroup -- identifies the container
-     COMM,     // contents of /proc/<pid>/comm
+     COMM,     // contents of /proc/<pid>/comm   -- identifies the program
      UID,
-     GID
+     GID,
+     CMDLINE,  // full /proc/<pid>/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
+  enum class Direction
-  {
+    {
-    Field       field;
+     READ,
-    Op          op;
+     WRITE
-    std::string pattern;
+    };
-    u32         number;      // parsed form of `pattern` for UID/GID
-    const Class *cls;        // owned by the RuleSet holding this rule
 
-    bool matches(const std::string &cgroup,
+  // Everything a rule can be matched against. Assembled once per
-                 const std::string &comm,
+  // request, and only for the fields some rule actually uses.
-                 const u32          uid,
+  struct Subject
-                 const u32          gid) const;
+  {
+    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<Condition> 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 Class *classify(const Subject &) const;
-                          const std::string &comm,
-                          const u32          uid,
-                          const u32          gid) 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<std::unique_ptr<Class>> &classes() const { return _classes; }
 
     std::string to_string() const;
 
   private:
-    std::vector<std::unique_ptr<Class>> _classes;
+    std::vector<std::unique_ptr<Class>>    _classes;
-    std::vector<Rule>                   _rules;
+    std::vector<Rule>                      _rules;
-    const Class                        *_default = nullptr;
+    std::unordered_map<std::string,u64>    _capacity;
-    bool                                _inert = true;
+    u64                                    _capacity_default = 0;
-    bool                                _needs_cgroup = false;
+    const Class                           *_default = nullptr;
-    bool                                _needs_comm = false;
+    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

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/<pid>/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}
   };

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 <path>=<mode>; 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()

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