This commit is contained in:
Grioghar Thomas-Baldwin 2026-10-03 18:58:58 -05:00 • committed by GitHub
commit 494faba737
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 3966 additions and 15 deletions

329
mkdocs/docs/config/qos.md Normal file
View file

@ -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.

View file

@ -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

View file

@ -117,6 +117,7 @@ Config::Config()
process_thread_count(fuse_cfg.process_thread_count),
process_thread_queue_depth(fuse_cfg.process_thread_queue_depth),
proxy_ioprio(false),
qos(false),
read_thread_count(fuse_cfg.read_thread_count),
readahead(0),
readdir("seq"),
@ -281,6 +282,14 @@ Config::Config()
_map["readahead"] = &readahead;
_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;
_map["qos.ruleset"] = &qos_ruleset;
_map["qos.stats"] = &qos_stats;
_map["rename-exdev"] = &rename_exdev;
_map["scheduling-priority"] = &scheduling_priority;
_map["security-capability"] = &security_capability;

View file

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

216
src/config_qos.cpp Normal file
View file

@ -0,0 +1,216 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include "config_qos.hpp"
#include "from_string.hpp"
#include "qos.hpp"
#include "syslog.hpp"
#include "fmt/core.h"
#include <errno.h>
#include <stdlib.h>
QoS::QoS(const bool b_)
{
qos::enable(b_);
}
std::string
QoS::to_string(void) const
{
return (qos::enabled() ? "true" : "false");
}
int
QoS::from_string(const std::string_view s_)
{
int rv;
bool enable;
rv = str::from(s_,&enable);
if(rv)
return rv;
qos::enable(enable);
return 0;
}
std::string
QoSRules::to_string(void) const
{
return qos::rules_path();
}
int
QoSRules::from_string(const std::string_view s_)
{
int rv;
std::string err;
const std::string path{s_};
rv = qos::load_file(path,&err);
if(rv < 0)
{
SysLog::error("qos.rules: {}: {}",path,err);
return rv;
}
SysLog::info("qos.rules: loaded {}",path);
return 0;
}
QoSRuleSet::QoSRuleSet()
{
ro = true;
}
std::string
QoSRuleSet::to_string(void) const
{
qos::RuleSet::Ptr rs = qos::ruleset();
if(rs == nullptr)
return {};
return rs->to_string();
}
int
QoSRuleSet::from_string(const std::string_view)
{
return -EINVAL;
}
std::string
QoSStats::to_string(void) const
{
return qos::stats();
}
int
QoSStats::from_string(const std::string_view s_)
{
if(s_ != "reset")
return -EINVAL;
qos::reset_stats();
return 0;
}
std::string
QoSMaxSleepers::to_string(void) const
{
return std::to_string(qos::max_sleepers.load(std::memory_order_relaxed));
}
int
QoSMaxSleepers::from_string(const std::string_view s_)
{
int rv;
int n;
rv = str::from(s_,&n);
if(rv)
return rv;
// -1 keeps the automatic sizing.
if(n < -1)
return -EINVAL;
qos::max_sleepers.store(n,std::memory_order_relaxed);
return 0;
}
std::string
QoSMaxSleepMS::to_string(void) const
{
return std::to_string(qos::max_sleep_ns.load(std::memory_order_relaxed) /
(1000 * 1000));
}
int
QoSMaxSleepMS::from_string(const std::string_view s_)
{
int rv;
int n;
rv = str::from(s_,&n);
if(rv)
return rv;
if(n < 0)
return -EINVAL;
qos::max_sleep_ns.store(static_cast<u64>(n) * 1000 * 1000,
std::memory_order_relaxed);
return 0;
}
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;
}

96
src/config_qos.hpp Normal file
View file

@ -0,0 +1,96 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#pragma once
#include "tofrom_string.hpp"
// qos=<bool>
class QoS : public ToFromString
{
public:
QoS(const bool);
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.rules=<filepath>
//
// Assigning a path loads it. Assigning the same path again is how a
// ruleset is reloaded at runtime. A file that fails to parse leaves
// the running ruleset in place.
class QoSRules : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.ruleset -- read only, the ruleset as parsed.
class QoSRuleSet : public ToFromString
{
public:
QoSRuleSet();
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.stats -- counters. Assigning "reset" zeroes them.
class QoSStats : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.max-sleepers=<int>
class QoSMaxSleepers : public ToFromString
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};
// qos.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
{
public:
std::string to_string(void) const final;
int from_string(const std::string_view) final;
};

View file

@ -22,6 +22,7 @@
#include "fileinfo.hpp"
#include "fs_pread.hpp"
#include "ioprio.hpp"
#include "qos.hpp"
#include "state.hpp"
#include "fuse.h"
@ -72,10 +73,25 @@ FUSE::read(const fuse_req_ctx_t *ctx_,
if(not fi)
return -EBADF;
if(fi->direct_io)
return ::_read_direct_io(fi->fd,buf_,size_,offset_);
// Classified after the file is known so path rules can see which
// file is being read, and so the rate is charged against the branch
// actually serving it.
qos::Apply q(ctx_,&fi->fusepath.native(),qos::Direction::READ);
qos::throttle(q,size_,fi->branch.path.native());
return ::_read_cached(fi->fd,buf_,size_,offset_);
// Timed only for protected classes. How long a player's reads
// actually take is the signal the governor throttles everything
// else against, so it has to be measured around the real I/O rather
// than inferred.
const u64 t0 = qos::timing_start(q);
const int rv = (fi->direct_io
? ::_read_direct_io(fi->fd,buf_,size_,offset_)
: ::_read_cached(fi->fd,buf_,size_,offset_));
qos::timing_end(q,fi->branch.path.native(),t0);
return rv;
}
int

View file

@ -27,6 +27,7 @@
#include "fs_pwrite.hpp"
#include "fs_pwriten.hpp"
#include "ioprio.hpp"
#include "qos.hpp"
#include "state.hpp"
#include "scope_guard/scope_guard.hpp"
@ -128,18 +129,11 @@ _move_and_pwriten(char const *buf_,
static
int
_write(const fuse_req_ctx_t *ctx_,
const fuse_file_info_t *ffi_,
const char *buf_,
const size_t count_,
const off_t offset_)
_write(FileInfo *fi,
const char *buf_,
const size_t count_,
const off_t offset_)
{
FileInfo *fi;
fi = state.get_fi(ctx_,ffi_->fh);
if(not fi)
return -EBADF;
// Concurrent writes can only happen if:
// 1) writeback-cache is enabled and using page caching
// 2) parallel_direct_writes is enabled and file has
@ -202,8 +196,26 @@ FUSE::write(const fuse_req_ctx_t *ctx_,
off_t offset_)
{
ioprio::SetFrom iop(ctx_->pid);
FileInfo *fi;
return ::_write(ctx_,ffi_,buf_,count_,offset_);
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,count_,fi->branch.path.native());
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

View file

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

View file

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

775
src/qos.cpp Normal file
View file

@ -0,0 +1,775 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include "qos.hpp"
#include "config.hpp"
#include "ioprio.hpp"
#include "procfs.hpp"
#include "fmt/core.h"
#include <errno.h>
#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
{
std::atomic<bool> _enabled{false};
// Deliberately small. Each sleeping process thread is one fewer
// thread draining the shared request queue, so the ceiling has to
// stay well under process-thread-count. See qos.hpp.
// -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
{
constexpr u64 NS_PER_SEC = 1000000000ULL;
// How long a pid's classification is trusted before it is looked up
// again. Short enough that a pid recycled onto a different class is
// corrected quickly, long enough that a streaming reader is not
// re-reading /proc on every request.
constexpr u64 CACHE_TTL_NS = 10 * NS_PER_SEC;
constexpr int CACHE_SIZE = 8;
// Ceiling on how far into the future a class's refill point may be
// pushed. Without it a burst that cannot be slept off -- because
// sleeps are capped, or too many threads are already sleeping --
// would leave the class paying down debt long after the burst
// ended, throttling traffic that did nothing wrong.
constexpr u64 MAX_DEBT_NS = 2 * NS_PER_SEC;
// 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};
std::string g_path;
// Function-local static so the first ruleset is built on demand
// rather than during static initialisation, where its ordering
// against Config's constructor would not be defined.
RuleSet::Ptr &
_ruleset()
{
static RuleSet::Ptr rs = RuleSet::make_default();
return rs;
}
struct CacheEntry
{
u32 pid = 0;
u64 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
// than copying it per request, is what makes the bare Class
// pointers in the cache safe: the ruleset a thread is using cannot
// be freed while that thread still references it, and the thread
// only swaps to a newer one at a generation change, which also
// clears the cache.
struct TLS
{
RuleSet::Ptr rs;
u64 generation = 0;
CacheEntry cache[CACHE_SIZE];
int next = 0;
int applied_ioprio = qos::UNSET;
int applied_nice = qos::UNSET;
};
thread_local TLS tls;
// 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()
{
struct timespec ts;
::clock_gettime(CLOCK_MONOTONIC,&ts);
return ((static_cast<u64>(ts.tv_sec) * NS_PER_SEC) +
static_cast<u64>(ts.tv_nsec));
}
void
_refresh_ruleset()
{
const u64 gen = g_generation.load(std::memory_order_acquire);
if(tls.generation == gen)
return;
{
std::lock_guard<std::mutex> lk(g_mutex);
tls.rs = ::_ruleset();
}
tls.generation = gen;
for(auto &e : tls.cache)
e = CacheEntry{};
tls.next = 0;
}
// The cache is keyed by the whole subject, not just the pid: with
// path rules in play the same process reading two different folders
// can land in two different classes.
bool
_lookup(const u32 pid_,
const u64 key_,
const u64 now_,
const Class **cls_)
{
for(const auto &e : tls.cache)
{
if(e.valid && (e.pid == pid_) && (e.key == key_) && (now_ < e.expire))
{
*cls_ = e.cls;
return true;
}
}
return false;
}
void
_remember(const u32 pid_,
const u64 key_,
const u64 now_,
const Class *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
qos::enable(const bool enable_)
{
_enabled.store(enable_,std::memory_order_relaxed);
}
RuleSet::Ptr
qos::ruleset()
{
std::lock_guard<std::mutex> lk(g_mutex);
return ::_ruleset();
}
void
qos::set_ruleset(RuleSet::Ptr rs_)
{
{
std::lock_guard<std::mutex> lk(g_mutex);
::_ruleset() = std::move(rs_);
}
// Published after the swap so a thread that sees the new generation
// is guaranteed to see the new ruleset.
g_generation.fetch_add(1,std::memory_order_release);
}
int
qos::load_file(const std::string &path_,
std::string *err_)
{
std::ifstream ifstrm;
ifstrm.open(path_);
if(!ifstrm.good())
{
*err_ = fmt::format("unable to open {}",path_);
return -EIO;
}
std::stringstream ss;
ss << ifstrm.rdbuf();
RuleSet::Ptr rs = RuleSet::parse(ss.str(),err_);
if(rs == nullptr)
return -EINVAL;
qos::set_ruleset(std::move(rs));
{
std::lock_guard<std::mutex> lk(g_mutex);
g_path = path_;
}
return 0;
}
u64
qos::timing_start(const Apply &apply_)
{
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;
const u64 now = ::_now_ns();
const u64 latency = ((now > started_) ? (now - started_) : 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(bucket->mutex);
const u64 now = ::_now_ns();
if(!bucket->primed)
{
bucket->tokens = burst;
bucket->refilled = now;
bucket->primed = true;
}
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 - bucket->refilled);
const unsigned __int128 added =
((elapsed * static_cast<unsigned __int128>(rate)) / NS_PER_SEC);
if(added >= burst)
bucket->tokens = burst;
else
bucket->tokens = std::min<u64>(burst,
bucket->tokens + static_cast<u64>(added));
bucket->refilled = now;
}
if(bucket->tokens >= bytes_)
{
bucket->tokens -= bytes_;
return;
}
const u64 deficit = (bytes_ - bucket->tokens);
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
// request that sleeps off its own deficit would otherwise find
// those same tokens waiting for it -- and hand them to the next
// request as well, which lets a class run at twice its rate.
//
// Moving `refilled` past the debt spends that time in advance. It
// also means a request allowed through unthrottled below still
// owes for its bytes; the next request in the class pays.
const u64 need_ns = static_cast<u64>((static_cast<unsigned __int128>(deficit) *
NS_PER_SEC) / rate);
bucket->refilled += need_ns;
if(bucket->refilled > (now + MAX_DEBT_NS))
bucket->refilled = (now + MAX_DEBT_NS);
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);
}
if(sleep_ns == 0)
return;
const int sleepers = (g_sleepers.fetch_add(1,std::memory_order_relaxed) + 1);
if(sleepers > ::_sleeper_limit())
{
g_sleepers.fetch_sub(1,std::memory_order_relaxed);
cls->passed.fetch_add(1,std::memory_order_relaxed);
return;
}
struct timespec ts;
ts.tv_sec = (sleep_ns / NS_PER_SEC);
ts.tv_nsec = (sleep_ns % NS_PER_SEC);
::nanosleep(&ts,nullptr);
g_sleepers.fetch_sub(1,std::memory_order_relaxed);
cls->throttled.fetch_add(1,std::memory_order_relaxed);
cls->throttled_ns.fetch_add(sleep_ns,std::memory_order_relaxed);
}
void
qos::Apply::_slow_apply(const fuse_req_ctx_t *ctx_,
const std::string *fusepath_,
const Direction dir_)
{
::_refresh_ruleset();
const RuleSet *rs = tls.rs.get();
if((rs == nullptr) || rs->inert())
return;
_rs = rs;
const u64 now = ::_now_ns();
const u64 key = ::_subject_key((rs->needs_path() ? fusepath_ : nullptr),dir_);
const Class *cls;
if(!::_lookup(ctx_->pid,key,now,&cls))
{
// Only the fields some rule actually tests are read, so a
// ruleset matching on cgroup alone costs one file read per
// cache miss rather than two.
const std::string cgroup = (rs->needs_cgroup()
? procfs::get_cgroup(ctx_->pid)
: std::string{});
const std::string comm = (rs->needs_comm()
? procfs::get_name(ctx_->pid)
: std::string{});
const std::string cmdline = (rs->needs_cmdline()
? procfs::get_cmdline(ctx_->pid)
: std::string{});
Subject subject;
subject.cgroup = &cgroup;
subject.comm = &comm;
subject.cmdline = &cmdline;
subject.path = fusepath_;
subject.uid = ctx_->uid;
subject.gid = ctx_->gid;
subject.dir = dir_;
cls = rs->classify(subject);
::_remember(ctx_->pid,key,now,cls);
}
if(cls == nullptr)
return;
_cls = cls;
if((cls->ioprio != qos::UNSET) && (cls->ioprio != tls.applied_ioprio))
{
if(::ioprio::set(0,cls->ioprio) >= 0)
tls.applied_ioprio = cls->ioprio;
}
if((cls->nice != qos::UNSET) && (cls->nice != tls.applied_nice))
{
errno = 0;
if((::setpriority(PRIO_PROCESS,0,cls->nice) == 0) || (errno == 0))
tls.applied_nice = cls->nice;
}
}
std::string
qos::rules_path()
{
std::lock_guard<std::mutex> lk(g_mutex);
return g_path;
}
std::string
qos::stats()
{
RuleSet::Ptr rs = qos::ruleset();
std::string out;
if(rs == nullptr)
return {};
out += fmt::format("enabled={} sleepers-in-flight={} max-sleepers={} "
"max-sleep-ms={}\n",
(qos::enabled() ? "true" : "false"),
g_sleepers.load(std::memory_order_relaxed),
::_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={} "
"throttled-ms={} passed={}\n",
c->name,
c->requests.load(std::memory_order_relaxed),
c->bytes.load(std::memory_order_relaxed),
c->throttled.load(std::memory_order_relaxed),
(c->throttled_ns.load(std::memory_order_relaxed) /
(1000 * 1000)),
c->passed.load(std::memory_order_relaxed));
}
return out;
}
void
qos::reset_stats()
{
RuleSet::Ptr rs = qos::ruleset();
if(rs == nullptr)
return;
for(const auto &c : rs->classes())
{
c->requests.store(0,std::memory_order_relaxed);
c->bytes.store(0,std::memory_order_relaxed);
c->throttled.store(0,std::memory_order_relaxed);
c->throttled_ns.store(0,std::memory_order_relaxed);
c->passed.store(0,std::memory_order_relaxed);
}
}

177
src/qos.hpp Normal file
View file

@ -0,0 +1,177 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
/*
Per-client quality of service.
mergerfs is a single daemon serving every process that touches the
pool, so from the kernel's point of view all pool I/O is issued by
one cgroup and one set of threads. Block layer priority
(ionice/BFQ, cgroup io.weight) therefore cannot tell a media player
apart from a torrent client: they are flattened into the same
issuer.
This restores the distinction inside the daemon. Each request is
attributed to the calling process, matched against a ruleset, and
the worker thread is given the ioprio and nice value of the class it
matched for the duration of the request. Optionally the class is
also rate limited.
Compared to the `proxy_ioprio` option, which copies whatever ioprio
the caller happens to have, the policy lives in one place and does
not depend on every client being correctly ioniced by whoever
started it -- which matters most for clients that respawn workers,
or that live in a container the daemon does not control.
*/
#pragma once
#include "qos_class.hpp"
#include "qos_rules.hpp"
#include "fuse_req_ctx.h"
#include <atomic>
#include <string>
namespace qos
{
// Exposed so the disabled-path check inlines into every
// fuse_read/fuse_write as an atomic load and a branch.
extern std::atomic<bool> _enabled;
[[gnu::always_inline]]
inline
bool
enabled()
{
return _enabled.load(std::memory_order_relaxed);
}
void enable(const bool);
// Replaces the active ruleset. Safe to call while requests are in
// flight.
void set_ruleset(RuleSet::Ptr);
RuleSet::Ptr ruleset();
// Loads and installs a ruleset from `path`. On failure the existing
// ruleset is left untouched and `err` describes the problem.
int load_file(const std::string &path, std::string *err);
// Number of process threads allowed to be sleeping in a throttle at
// once, and the longest any single request may be delayed.
//
// Both exist to bound the damage throttling can do. mergerfs hands
// requests from its read threads to a *bounded* process thread
// queue, so a sleeping process thread is a process thread not
// serving anyone -- and once the queue backs up, requests of every
// class queue behind it. Rather than let a rate limit turn into a
// pool-wide stall, a request that would exceed either bound is
// allowed through unthrottled and counted in the `passed` stat.
extern std::atomic<int> max_sleepers;
extern std::atomic<u64> max_sleep_ns;
// 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();
// Feedback loop.
//
// A protected class's service time is the control signal. When reads
// for a player start taking materially longer than that same disk
// manages when it is quiet, pressure on the resource rises and every
// yielding class's allowance shrinks in proportion to its `yield`
// rank -- downloads first and hardest, a player's own background
// work more gently, playback not at all. When the latency recovers,
// or playback stops altogether, pressure decays and the allowances
// return to full.
//
// Returns 0 for a request whose class is not protected, in which
// case timing_end does nothing. Keeping the clock read out of the
// unprotected path is why this is split in two.
u64 timing_start(const Apply &);
void timing_end(const Apply &, const std::string &resource, const u64 started);
// Current backoff for a resource, 0.0 to 1.0. Zero when no protected
// class has touched it recently, which is what lets bulk traffic run
// flat out on a disk nobody is streaming from.
double pressure(const std::string &resource);
// Records that a class willing to yield has just issued I/O against
// a resource, which is what makes it count as contended.
void note_yielding(const std::string &resource);
// Charges `bytes` to the class and, if that class is over its rate
// for `resource`, sleeps for as long as the bounds above permit.
// Called with the size the caller asked for, before the I/O is
// issued.
//
// `resource` is the branch serving the request, so a rate is a
// per-device allowance: a class limited to 10% does not have to
// share one budget across seven disks.
void throttle(const Apply &, const u64 bytes, const std::string &resource);
// Classifies the calling process and applies its class to this
// thread. The thread keeps those settings until another request
// changes them, so nothing is restored on destruction -- FUSE
// worker threads do nothing between requests worth protecting, and
// restoring would double the syscalls on the hot path.
class Apply
{
public:
[[gnu::always_inline]]
inline
Apply(const fuse_req_ctx_t *ctx_,
const std::string *fusepath_,
const Direction dir_)
{
if(qos::enabled())
_slow_apply(ctx_,fusepath_,dir_);
}
// nullptr when QoS is off or the ruleset cannot change anything.
const Class *cls() const { return _cls; }
const RuleSet *ruleset() const { return _rs; }
private:
void _slow_apply(const fuse_req_ctx_t *,
const std::string *,
const Direction);
private:
const Class *_cls = nullptr;
const RuleSet *_rs = nullptr;
};
}

37
src/qos_class.cpp Normal file
View file

@ -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();
}

207
src/qos_class.hpp Normal file
View file

@ -0,0 +1,207 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#pragma once
#include "base_types.h"
#include <atomic>
#include <mutex>
#include <string>
#include <memory>
#include <unordered_map>
namespace qos
{
// Sentinel meaning "leave the worker thread's current value alone".
constexpr int UNSET = INT32_MIN;
namespace ioprio
{
// Mirrors the kernel's linux/ioprio.h encoding. Redefined here so
// mergerfs does not depend on a UAPI header that has moved
// between kernel versions.
constexpr int CLASS_SHIFT = 13;
constexpr int PRIO_MASK = (1 << CLASS_SHIFT) - 1;
constexpr int CLASS_NONE = 0;
constexpr int CLASS_RT = 1;
constexpr int CLASS_BE = 2;
constexpr int CLASS_IDLE = 3;
constexpr
int
value(const int class_,
const int data_)
{
return ((class_ << CLASS_SHIFT) | data_);
}
constexpr
int
to_class(const int value_)
{
return (value_ >> CLASS_SHIFT);
}
constexpr
int
to_data(const int value_)
{
return (value_ & PRIO_MASK);
}
std::string to_string(const int value);
// Parses "rt:N", "be:N", "idle" and "none". Returns 0 on success
// and -EINVAL otherwise.
int from_string(const std::string_view, int *value);
}
// 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 buckets and counters belonging to every request matched to
// it.
//
// The descriptive members are immutable once a ruleset is
// published, which is what lets the classifier hand out bare
// pointers to it. Everything mutable is either atomic or guarded.
class Class
{
public:
Class(std::string name_)
: name(std::move(name_))
{
}
public:
std::string name;
int ioprio = qos::UNSET;
int nice = qos::UNSET;
// A rate is either absolute (`rate` bytes/sec) or a share of the
// serving resource's measured capacity (`pct` percent). Both zero
// means unlimited.
u64 rate = 0;
u32 pct = 0;
// Bucket depth. Zero means "one second of whatever the rate
// resolves to", which cannot be precomputed for a percentage.
u64 burst = 0;
// Latency of this class's requests is the signal the governor
// controls against, and it is never itself throttled.
bool protect = false;
// Never delayed, and never a control signal.
//
// This exists for the services playback *synchronously waits on*
// rather than playback itself -- Plex's EasyAudioEncoder is the
// canonical case: a transcode of EAC3/TrueHD/DTS audio writes to
// EAE and blocks until it answers. Slowing such a helper slows the
// stream that is waiting on it, so the "bulk" it appears to be
// doing must be left alone. Getting this wrong is a priority
// inversion: the machinery meant to protect playback stalls it.
bool critical = false;
// True when this class must never be delayed for any reason.
bool immune() const { return (protect || critical); }
// How strongly this class gives way as pressure rises, 0-100.
// 0 never yields; 100 yields its whole allowance at full
// pressure. This is the priority ladder: downloads yield hardest,
// a player's background work yields some, playback yields none.
u32 yield = 0;
// Never back off below this, so a yielding class is slowed rather
// than stopped. Either absolute bytes/sec or a percentage of the
// resource's capacity.
u64 floor = 0;
u32 floor_pct = 0;
bool limited() const { return ((rate != 0) || (pct != 0)); }
bool adaptive() const { return (yield != 0); }
public:
// Buckets are created on first use and keyed by resource name
// (a branch path, or "pool" when the branch is unknown). Seven
// branches means seven entries, so a plain map under a mutex is
// cheaper than anything cleverer.
mutable std::mutex buckets_mutex;
mutable std::unordered_map<std::string,
std::unique_ptr<Bucket>> buckets;
Bucket *bucket_for(const std::string &resource) const;
public:
// Counters, reported via the qos.stats config key.
mutable std::atomic<u64> requests{0};
mutable std::atomic<u64> bytes{0};
mutable std::atomic<u64> throttled{0};
mutable std::atomic<u64> throttled_ns{0};
mutable std::atomic<u64> passed{0};
};
}

720
src/qos_rules.cpp Normal file
View file

@ -0,0 +1,720 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include "qos_rules.hpp"
#include "fmt/core.h"
#include <errno.h>
#include <fnmatch.h>
#include <cctype>
#include <cstdlib>
#include <sstream>
using qos::Class;
using qos::Condition;
using qos::Direction;
using qos::Field;
using qos::MatchOp;
using qos::Rule;
using qos::RuleSet;
using qos::Subject;
std::string
qos::ioprio::to_string(const int value_)
{
if(value_ == qos::UNSET)
return "unset";
switch(qos::ioprio::to_class(value_))
{
case qos::ioprio::CLASS_NONE:
return "none";
case qos::ioprio::CLASS_RT:
return fmt::format("rt:{}",qos::ioprio::to_data(value_));
case qos::ioprio::CLASS_BE:
return fmt::format("be:{}",qos::ioprio::to_data(value_));
case qos::ioprio::CLASS_IDLE:
return "idle";
}
return "unset";
}
int
qos::ioprio::from_string(const std::string_view s_,
int *value_)
{
if(s_ == "idle")
{
// The idle class carries no priority data.
*value_ = qos::ioprio::value(qos::ioprio::CLASS_IDLE,0);
return 0;
}
if(s_ == "none")
{
*value_ = qos::ioprio::value(qos::ioprio::CLASS_NONE,0);
return 0;
}
int klass;
if(s_.rfind("rt:",0) == 0)
klass = qos::ioprio::CLASS_RT;
else if(s_.rfind("be:",0) == 0)
klass = qos::ioprio::CLASS_BE;
else
return -EINVAL;
const std::string_view data = s_.substr(3);
if((data.size() != 1) || !std::isdigit(static_cast<unsigned char>(data[0])))
return -EINVAL;
const int n = (data[0] - '0');
if(n > 7)
return -EINVAL;
*value_ = qos::ioprio::value(klass,n);
return 0;
}
int
qos::parse_size(const std::string_view s_,
u64 *bytes_)
{
if(s_.empty())
return -EINVAL;
std::size_t i = 0;
while((i < s_.size()) &&
(std::isdigit(static_cast<unsigned char>(s_[i])) || (s_[i] == '.')))
i++;
if(i == 0)
return -EINVAL;
const std::string num(s_.substr(0,i));
errno = 0;
char *end = nullptr;
const double value = ::strtod(num.c_str(),&end);
if(errno || (end == num.c_str()) || (*end != '\0') || (value < 0))
return -EINVAL;
std::string suffix(s_.substr(i));
// Accept "MB" and "MiB" as aliases of "M". mergerfs treats all size
// suffixes as binary multiples, matching minfreespace and friends.
if((suffix.size() > 1) && (suffix.back() == 'B'))
suffix.pop_back();
if((suffix.size() > 1) && (suffix.back() == 'i'))
suffix.pop_back();
u64 mult;
if(suffix.empty())
mult = 1ULL;
else if((suffix == "K") || (suffix == "k"))
mult = 1024ULL;
else if((suffix == "M") || (suffix == "m"))
mult = 1024ULL * 1024;
else if((suffix == "G") || (suffix == "g"))
mult = 1024ULL * 1024 * 1024;
else if((suffix == "T") || (suffix == "t"))
mult = 1024ULL * 1024 * 1024 * 1024;
else
return -EINVAL;
*bytes_ = static_cast<u64>(value * static_cast<double>(mult));
return 0;
}
bool
qos::Condition::matches(const Subject &s_) const
{
const std::string *subject;
switch(field)
{
case Field::UID:
return (s_.uid == number);
case Field::GID:
return (s_.gid == number);
case Field::OP:
return (s_.dir == dir);
case Field::CGROUP:
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;
}
if(subject == nullptr)
return false;
if(op == MatchOp::EQ)
return (*subject == pattern);
// 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 *
RuleSet::find_class(const std::string_view name_) const
{
for(const auto &c : _classes)
{
if(c->name == name_)
return c.get();
}
return nullptr;
}
const Class *
RuleSet::classify(const Subject &s_) const
{
for(const auto &rule : _rules)
{
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()
{
auto rs = std::make_shared<RuleSet>();
rs->_classes.push_back(std::make_unique<Class>("default"));
rs->_default = rs->_classes.back().get();
rs->_inert = true;
return rs;
}
namespace
{
// 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::string tok;
bool in_tok = false;
char quote = '\0';
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;
}
int
parse_field(const std::string &s_,
Field *field_)
{
if(s_ == "cgroup")
*field_ = Field::CGROUP;
else if(s_ == "comm")
*field_ = Field::COMM;
else if(s_ == "uid")
*field_ = Field::UID;
else if(s_ == "gid")
*field_ = Field::GID;
else if(s_ == "cmdline")
*field_ = Field::CMDLINE;
else if(s_ == "path")
*field_ = Field::PATH;
else if(s_ == "op")
*field_ = Field::OP;
else
return -EINVAL;
return 0;
}
}
RuleSet::Ptr
RuleSet::parse(const std::string_view text_,
std::string *err_)
{
auto rs = std::make_shared<RuleSet>();
std::istringstream istrm{std::string(text_)};
std::string line;
std::string default_name;
int lineno = 0;
auto fail =
[&](const std::string &msg_) -> RuleSet::Ptr
{
*err_ = fmt::format("line {}: {}",lineno,msg_);
return nullptr;
};
while(std::getline(istrm,line,'\n'))
{
lineno++;
const std::size_t hash = line.find('#');
if(hash != std::string::npos)
line.erase(hash);
auto tokens = ::tokenize(line);
if(tokens.empty())
continue;
if(tokens[0] == "class")
{
if(tokens.size() < 2)
return fail("'class' needs a name");
if(rs->find_class(tokens[1]))
return fail(fmt::format("duplicate class '{}'",tokens[1]));
auto cls = std::make_unique<Class>(tokens[1]);
for(std::size_t i = 2; i < tokens.size(); i++)
{
// `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]));
const std::string key = tokens[i].substr(0,eq);
const std::string val = tokens[i].substr(eq + 1);
if(key == "ioprio")
{
if(qos::ioprio::from_string(val,&cls->ioprio))
return fail(fmt::format("invalid ioprio '{}'",val));
}
else if(key == "nice")
{
const int n = std::atoi(val.c_str());
if((n < -20) || (n > 19))
return fail(fmt::format("nice out of range: '{}'",val));
cls->nice = n;
}
else if(key == "rate")
{
if(!val.empty() && (val.back() == '%'))
{
const std::string num = val.substr(0,val.size() - 1);
char *end = nullptr;
const unsigned long pct = ::strtoul(num.c_str(),&end,10);
if((end == num.c_str()) || (*end != '\0') ||
(pct == 0) || (pct > 100))
return fail(fmt::format("percentage must be 1-100: '{}'",val));
cls->pct = static_cast<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));
}
}
// 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. For
// a percentage rate the depth can only be resolved once the
// serving resource is known, so it is left at zero here.
if(cls->rate && (cls->burst < cls->rate))
cls->burst = cls->rate;
if(cls->protect)
rs->_has_protected = true;
if((cls->ioprio != qos::UNSET) ||
(cls->nice != qos::UNSET) ||
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")
{
Rule rule{};
std::size_t i = 1;
// Conditions are (field, operator, pattern) triples running
// up to the arrow. Every one must match for the rule to
// fire.
while((i < tokens.size()) && (tokens[i] != "->"))
{
if((i + 2) >= tokens.size())
return fail("expected: match <field> <=|~> <pattern> ... -> <class>");
Condition cond{};
if(::parse_field(tokens[i],&cond.field))
return fail(fmt::format("unknown field '{}'",tokens[i]));
if(tokens[i+1] == "=")
cond.op = MatchOp::EQ;
else if(tokens[i+1] == "~")
cond.op = MatchOp::GLOB;
else
return fail(fmt::format("unknown operator '{}'",tokens[i+1]));
cond.pattern = tokens[i+2];
switch(cond.field)
{
case Field::UID:
case Field::GID:
{
if(cond.op != MatchOp::EQ)
return fail("uid/gid only support the '=' operator");
char *end = nullptr;
const unsigned long n = ::strtoul(cond.pattern.c_str(),&end,10);
if((end == cond.pattern.c_str()) || (*end != '\0'))
return fail(fmt::format("invalid uid/gid '{}'",cond.pattern));
cond.number = static_cast<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;
}
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[i+1]);
if(rule.cls == nullptr)
return fail(fmt::format("unknown class '{}' (classes must be "
"defined before use)",tokens[i+1]));
rs->_rules.push_back(std::move(rule));
}
else if(tokens[0] == "default")
{
if(tokens.size() != 2)
return fail("expected: default <class>");
default_name = tokens[1];
}
else
{
return fail(fmt::format("unknown directive '{}'",tokens[0]));
}
}
if(default_name.empty())
{
// No explicit default: unmatched processes keep whatever
// priority they already had.
rs->_classes.push_back(std::make_unique<Class>("unclassified"));
rs->_default = rs->_classes.back().get();
}
else
{
rs->_default = rs->find_class(default_name);
if(rs->_default == nullptr)
{
*err_ = fmt::format("default names unknown class '{}'",default_name);
return nullptr;
}
}
if(rs->_rules.empty() &&
(rs->_default->ioprio == qos::UNSET) &&
(rs->_default->nice == qos::UNSET) &&
!rs->_default->limited() &&
!rs->_default->adaptive() &&
!rs->_default->protect &&
!rs->_default->critical)
rs->_inert = true;
return rs;
}
std::string
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={} "
"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->pct ? fmt::format("{}%",c->pct) : std::to_string(c->rate)),
c->burst,
c->yield,
(c->floor_pct ? fmt::format("{}%",c->floor_pct)
: std::to_string(c->floor)));
}
for(const auto &r : _rules)
{
s += "match";
for(const auto &cond : r.conditions)
{
const char *field = "";
switch(cond.field)
{
case Field::CGROUP: field = "cgroup"; break;
case Field::COMM: field = "comm"; break;
case Field::CMDLINE: field = "cmdline"; break;
case Field::UID: field = "uid"; break;
case Field::GID: field = "gid"; break;
case Field::PATH: field = "path"; break;
case Field::OP: field = "op"; break;
}
s += fmt::format(" {} {} {}",
field,
((cond.op == MatchOp::EQ) ? "=" : "~"),
cond.pattern);
}
s += fmt::format(" -> {}\n",r.cls->name);
}
if(_default)
s += fmt::format("default {}\n",_default->name);
return s;
}

168
src/qos_rules.hpp Normal file
View file

@ -0,0 +1,168 @@
/*
ISC License
Copyright (c) 2026, Antonio SJ Musumeci <trapexit@spawn.link>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#pragma once
#include "qos_class.hpp"
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
namespace qos
{
// Which attribute of a request a condition tests.
enum class Field
{
CGROUP, // contents of /proc/<pid>/cgroup -- identifies the container
COMM, // contents of /proc/<pid>/comm -- identifies the program
UID,
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 MatchOp
{
EQ, // '=' exact string / numeric equality
GLOB // '~' fnmatch(3) pattern, '/' not treated specially
};
enum class Direction
{
READ,
WRITE
};
// Everything a rule can be matched against. Assembled once per
// request, and only for the fields some rule actually uses.
struct Subject
{
const std::string *cgroup = nullptr;
const std::string *comm = nullptr;
const std::string *cmdline = nullptr;
const std::string *path = nullptr;
u32 uid = 0;
u32 gid = 0;
Direction dir = Direction::READ;
};
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
// request in flight. Readers keep a shared_ptr for the duration of
// a request, which is what keeps a Class alive after a swap.
class RuleSet
{
public:
typedef std::shared_ptr<const RuleSet> Ptr;
public:
// Builds the ruleset every mount starts with: a single
// pass-through class and no rules, which leaves behaviour
// identical to qos being off.
static Ptr make_default();
// Parses `text` and returns the new ruleset, or nullptr on
// error, in which case `err` describes the first problem found
// and the caller must keep using the ruleset it already has.
static Ptr parse(const std::string_view text,
std::string *err);
public:
const Class *classify(const Subject &) const;
const Class *find_class(const std::string_view name) const;
// Measured or declared throughput of `resource` in bytes/sec,
// falling back to the `capacity default` line. Zero when nothing
// is known, which makes percentage rates unenforceable and is
// reported as such.
u64 capacity(const std::string &resource) const;
// Resolves a class's configured rate against the resource
// serving the request. Returns 0 for "unlimited".
u64 rate_for(const Class *, const std::string &resource) const;
// True when no rule and no class can change a thread's behaviour,
// letting the hot path skip reading /proc entirely.
bool inert() const { return _inert; }
// Which subject fields any rule actually inspects. Each one that
// is unused is a file read or a string copy skipped per request.
bool needs_cgroup() const { return _needs_cgroup; }
bool needs_comm() const { return _needs_comm; }
bool needs_cmdline() const { return _needs_cmdline; }
bool needs_path() const { return _needs_path; }
// True when some class is marked `protect`, which is what turns
// the adaptive governor on.
bool has_protected() const { return _has_protected; }
// Resolves a class's floor against the resource.
u64 floor_for(const Class *, const std::string &resource) const;
const std::vector<std::unique_ptr<Class>> &classes() const { return _classes; }
std::string to_string() const;
private:
std::vector<std::unique_ptr<Class>> _classes;
std::vector<Rule> _rules;
std::unordered_map<std::string,u64> _capacity;
u64 _capacity_default = 0;
const Class *_default = nullptr;
bool _inert = true;
bool _needs_cgroup = false;
bool _needs_comm = false;
bool _needs_cmdline = false;
bool _needs_path = false;
bool _has_protected = false;
};
// Parses "10M", "1.5MiB", "512K", "1G" and bare byte counts into
// bytes. Returns 0 on success, -EINVAL otherwise.
int parse_size(const std::string_view, u64 *bytes);
}

View file

@ -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}
};

314
tools/mergerfs.qos-bench Executable file
View file

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

324
tools/mergerfs.qos-playback-test Executable file
View file

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