elars/scx_soryu
1
0 Fork 0
mirror of https://github.com/elwrcl/scx_soryu.git synced 2026-10-31 02:48:03 +03:00
Code Issues Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
elwrcl committed 2026-09-02 20:00:01 +03:00
commit b64e40d88d
12 files changed
+3566
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+6
View File
@@ -0,0 +1,6 @@
/target/
/result
/result-*
src/bpf/vmlinux.h
/nixos.qcow2
/Cargo.lock.orig
Cargo.lock
Generated
+1540
View File
File diff suppressed because it is too large. Load diff
Cargo.toml
+27
View File
@@ -0,0 +1,27 @@
[package]
name = "scx_soryu"
version = "0.1.0"
edition = "2021"
description = "BMQ policy port for sched_ext"
license = "GPL-2.0-only"
[dependencies]
anyhow = "1.0"
clap = { version = "4", features = ["derive", "env", "wrap_help"] }
ctrlc = { version = "3.4", features = ["termination"] }
libbpf-rs = "=0.26.2"
libc = "0.2"
log = "0.4"
scx_stats = "1.1.3"
scx_stats_derive = "1.1.3"
scx_utils = "1.1.3"
serde = { version = "1", features = ["derive"] }
crossbeam = "0.8"
simplelog = "0.12"
[build-dependencies]
scx_cargo = "1.1.3"
[profile.release]
lto = "thin"
panic = "abort"
README.md
+38
View File
@@ -0,0 +1,38 @@
# scx_soryu
a port of the BMQ/LFBMQ scheduling policy onto [sched_ext](https://docs.kernel.org/scheduler/sched-ext.html).
the policy is pure BPF C; userspace is a thin Rust loader and more safer than normal scheduler.
## tested on
| machine | cpu | status |
| --- | --- | --- |
| target, tested | Intel i5-3210m (Ivy Bridge, 2c/4t) | **verified** — bare metal, extended game + build sessions |
| target, untested | Ryzen 5 PRO 4650u (Renoir, 6c/12t) | not yet run |
| target, untested | Ryzen 5 7500F (ZEN4, 6c/12t) | not yet run |
## development
```sh
nix develop # clang, bpftool, rust, libbpf; kernel drift guard
```
run it by hand:
```sh
sudo ./target/release/scx_soryu
```
or as a NixOS service — see [`docs/service.md`](docs/service.md).
test VM (a smoke test, not a benchmark — measure on bare metal):
```sh
nix build .#nixosConfigurations.testvm.config.system.build.vm
./result/bin/run-nixos-vm
```
## kernel
audited series is **7.1.x**. the devShell refuses to open outside it.
`vmlinux.h` is generated from the running kernel's BTF and is not committed.
build.rs
+7
View File
@@ -0,0 +1,7 @@
fn main() {
scx_cargo::BpfBuilder::new()
.unwrap()
.enable_skel("src/bpf/main.bpf.c", "bpf")
.build()
.unwrap();
}
docs/service.md
+99
View File
@@ -0,0 +1,99 @@
# the NixOS service
the scheduler is now also a flake package and a NixOS module. running
`sudo ./target/release/scx_soryu` by hand still works — the service is an
addition, not a replacement.
## adding it to copland
`flake.nix` inputs:
```nix
scx-soryu.url = "github:elwrcl/scx_soryu/bmq";
```
in the configuration:
```nix
imports = [ inputs.scx-soryu.nixosModules.default ];
services.scx-soryu = {
enable = true;
notify.user = "elars"; # desktop notification on ejection
};
```
that's it. the defaults are measured values: `sliceUs = 1000`
([`slice.md`](slice.md)), `starveMs = 128` ([`stage6-starve.md`](stage6-starve.md)),
`timeoutMs = 5000`, `reportSec = 60`.
to turn it off: `services.scx-soryu.enable = false;` or
`systemctl stop scx-soryu`. either one falls back to EEVDF immediately.
## "something happened and nobody noticed"
an ejection is a silent event: the kernel drops the scheduler, every task
falls back to default kernel scheduler, and nothing on the desktop says so. three layers were added to fix that.
**1. the loader says so, loudly.** ejection used to be a bare
`Error: ...` line printed by anyhow — easy to scroll past in a terminal,
unlabeled and unsearchable in the journal. now it's two ERROR-level lines:
```
scx_soryu was ejected by the kernel: EXIT: runnable task stall (...)
all tasks are back on default kernel scheduler; the scheduler is NOT running
```
**2. systemd retries three times, then gives up.** `Restart=on-failure`,
3 starts in 5 minutes. a transient load spike recovers on its own; a
persistent problem doesn't flap forever — it lands in `failed`.
**3. `failed` triggers a desktop notification.** a second unit bound with
`OnFailure=` finds the user's session bus and calls
`notify-send --urgency=critical`. if there's no session, it exits quietly
(no notification, no error either).
either way, the full story is in the journal:
```
journalctl -u scx-soryu -n 200
```
## telemetry noise
the loader prints four lines a second — fine in a terminal, a disaster in
the journal. `--report-sec` was added; the service uses 60. shutdown is
still checked once a second regardless, so `systemctl stop` isn't slowed
down.
to see live status while the service is running:
```
sudo scx_soryu --monitor 3
```
## verified in a VM
`nixosConfigurations.servicevm` — the same test VM, but running the
scheduler as a service instead of by hand. `tools/servicevm.sh`:
| check | result |
| --- | --- |
| comes up at boot | `active`, `sched_ext=enabled` |
| clean `systemctl stop` | `inactive`, `sched_ext=disabled`, **not** `failed` |
| restarts | `active`, `enabled` |
| repeated SIGKILL | 2 restarts, then `failed` — no flapping |
| journal | `Start request repeated too quickly` + `Failed with result 'start-limit-hit'` |
a clean `systemctl stop` mattered: SIGTERM reaches the loader and drops the
struct_ops link, so unload is clean and the unit isn't marked `failed`.
## packaging note
`packages.default` builds inside the sandbox: the BPF object is compiled
against `scx_utils`'s vendored `vmlinux.h`, then CO-RE-relocated against
the running kernel's BTF at load time. so the build doesn't depend on the
running kernel — `/sys/kernel/btf/vmlinux` is only needed at **run** time.
(the `src/bpf/vmlinux.h` sitting in the repo turned out unused — deleting
it, the build still passes the same way. it was already gitignored.)
flake.lock
Generated
+61
View File
@@ -0,0 +1,61 @@
{
"nodes": {
"flake-utils": {
"inputs": {
"systems": "systems"
},
"locked": {
"lastModified": 1731533236,
"narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1784007870,
"narHash": "sha256-djcLt/JJphyNt4eDY9XTly+/WbCK5lqWq9lSgCmJkkQ=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "18b9261cb3294b6d2a06d03f96872827b8fe2698",
"type": "github"
},
"original": {
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "18b9261cb3294b6d2a06d03f96872827b8fe2698",
"type": "github"
}
},
"root": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
}
},
"systems": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
}
},
"root": "root",
"version": 7
}
flake.nix
+331
View File
@@ -0,0 +1,331 @@
{
description = "scx_soryu — BMQ/LFBMQ policy port for sched_ext";
# The test VM's kernel is the same derivation as
# the host's booted kernel. See docs/KERNEL.md.
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/18b9261cb3294b6d2a06d03f96872827b8fe2698";
flake-utils.url = "github:numtide/flake-utils";
};
outputs = { self, nixpkgs, flake-utils }:
let
system = "x86_64-linux";
pkgs = nixpkgs.legacyPackages.${system};
kernelPackages = pkgs.linuxPackages_7_1;
kernelVersion = kernelPackages.kernel.modDirVersion;
auditedSeries = "7.1";
vmBase = { pkgs, lib, ... }: {
boot.kernelPackages = kernelPackages;
# SysRq-S (sched_ext reset) and SysRq-D (dump) are gated behind
# SYSRQ_ENABLE_RTNICE (0x100). NixOS's default of 184 omits it, so
# the emergency escape hatch is dead unless we set this.
boot.kernel.sysctl."kernel.sysrq" = 440;
virtualisation = {
cores = 4;
memorySize = 4096;
graphics = false;
# the host: 2 physical cores x 2 threads, one LLC. Without
# this QEMU reports 4 sockets and scx_bpf_select_cpu_dfl()'s SMT
# walk has nothing real to walk.
qemu.options = [ "-cpu host" "-smp 4,sockets=1,cores=2,threads=2" ];
sharedDirectories.soryu = {
source = "/home/elars/Projects/scx_soryu";
target = "/mnt/soryu";
};
};
services.getty.autologinUser = "root";
users.users.root.password = "";
# rt-tests carries both cyclictest and hackbench (M1 harness).
environment.systemPackages = with pkgs; [ stress-ng rt-tests ];
system.stateVersion = "25.05";
};
in
{
devShells.${system}.default = pkgs.mkShell {
name = "scx_soryu";
nativeBuildInputs = with pkgs; [
clang
llvm
bpftools
pkg-config
rustc
cargo
rustfmt
clippy
rust-analyzer
];
buildInputs = with pkgs; [
libbpf
elfutils
zlib
zstd
];
# scx_utils' build script runs bindgen, which dlopens libclang.
LIBCLANG_PATH = "${pkgs.libclang.lib}/lib";
# The nix cc-wrapper injects -fzero-call-used-regs, which clang rejects
# for -target bpf. nixpkgs' own scx derivation disables the same knob.
hardeningDisable = [ "zerocallusedregs" ];
RUSTFLAGS = builtins.concatStringsSep " " [
"-C relocation-model=pic"
"-C link-args=-lelf"
"-C link-args=-lz"
"-C link-args=-lzstd"
];
shellHook = ''
expected="${kernelVersion}"
series="${auditedSeries}"
running="$(uname -r)"
running_series="$(echo "$running" | cut -d. -f1,2)"
if [ "$running_series" != "$series" ]; then
cat >&2 <<MSG
╔══════════════════════════════════════════════════════════════╗
║ KERNEL SERIES MISMATCH — refusing to enter ║
╚══════════════════════════════════════════════════════════════╝
audited series : $series.x
booted : $running
MSG
return 1
fi
if [ "$running" != "$expected" ]; then
cat >&2 <<MSG
warning: patch-level kernel drift
flake VM kernel : $expected
booted : $running
Same $series.x series, so the kfunc audit still holds. The test VM
will run $expected while bare metal runs $running. The BPF object
is compiled against scx_utils' vendored vmlinux.h and relocated
against the booted kernel's BTF at load time (CO-RE), so the build
does not depend on which of the two you are on.
MSG
fi
# Needed to *load*, not to build: libbpf resolves CO-RE relocations
# against this at attach time.
if [ ! -r /sys/kernel/btf/vmlinux ]; then
echo "error: /sys/kernel/btf/vmlinux unreadable; CONFIG_DEBUG_INFO_BTF required" >&2
return 1
fi
echo "scx_soryu devShell — kernel $running, BTF ok"
'';
};
packages.${system}.default = pkgs.rustPlatform.buildRustPackage {
pname = "scx_soryu";
version = "0.1.0";
# target/ is a gigabyte and nixos.qcow2 is the VM's disk; neither
# belongs in the store path, and both would invalidate the hash on
# every build.
src = pkgs.lib.cleanSourceWith {
src = ./.;
filter = path: _type:
!(builtins.elem (baseNameOf path) [ "target" "result" "nixos.qcow2" ]);
};
cargoLock.lockFile = ./Cargo.lock;
nativeBuildInputs = with pkgs; [
clang
llvm
bpftools
pkg-config
rustPlatform.bindgenHook # scx_utils runs bindgen; needs libclang
];
buildInputs = with pkgs; [ libbpf elfutils zlib zstd ];
# Same two workarounds the devShell needs, for the same reasons.
hardeningDisable = [ "zerocallusedregs" ];
RUSTFLAGS = builtins.concatStringsSep " " [
"-C relocation-model=pic"
"-C link-args=-lelf"
"-C link-args=-lz"
"-C link-args=-lzstd"
];
doCheck = false; # the tests that matter need a kernel, not a sandbox
meta = {
description = "BMQ/LFBMQ policy port for sched_ext";
mainProgram = "scx_soryu";
};
};
nixosModules.default = { config, lib, pkgs, ... }:
let
cfg = config.services.scx-soryu;
in
{
options.services.scx-soryu = {
enable = lib.mkEnableOption "the scx_soryu sched_ext scheduler";
package = lib.mkOption {
type = lib.types.package;
default = self.packages.${pkgs.stdenv.hostPlatform.system}.default;
description = "scx_soryu build to run.";
};
sliceUs = lib.mkOption {
type = lib.types.ints.positive;
default = 1000;
description = ''
Timeslice in microseconds. 1000 is the measured knee: see
docs/slice.md. 4194 restores BMQ's own value.
'';
};
starveMs = lib.mkOption {
type = lib.types.ints.unsigned;
default = 128;
description = ''
Anti-starvation deadline. 0 restores BMQ's semantics.
'';
};
timeoutMs = lib.mkOption {
type = lib.types.ints.positive;
default = 5000;
description = "Kernel watchdog threshold for a stalled task.";
};
reportSec = lib.mkOption {
type = lib.types.ints.positive;
default = 60;
description = ''
Seconds between telemetry lines in the journal. The loader's
own default is 1, which is right for a terminal and far too
chatty for a service.
'';
};
extraArgs = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [ ];
example = [ "--boost-on-wake" "false" ];
};
notify = {
enable = lib.mkOption {
type = lib.types.bool;
default = cfg.notify.user != null;
defaultText = lib.literalExpression "notify.user != null";
description = ''
Send a desktop notification when the scheduler gives up.
Being ejected is silent: the kernel puts every task back on
default scheduler and nothing on the desktop says so.
Without this the only trace is the journal.
'';
};
user = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;
example = "elars";
description = "Desktop user to notify.";
};
};
};
config = lib.mkIf cfg.enable {
assertions = [{
assertion = !cfg.notify.enable || cfg.notify.user != null;
message = "services.scx-soryu.notify.enable needs notify.user.";
}];
systemd.services.scx-soryu = {
description = "scx_soryu sched_ext scheduler";
documentation = [ "https://github.com/elwrcl/scx_soryu" ];
wantedBy = [ "multi-user.target" ];
after = [ "systemd-udev-settle.service" ];
startLimitIntervalSec = 300;
startLimitBurst = 3;
onFailure = lib.mkIf cfg.notify.enable [ "scx-soryu-notify.service" ];
serviceConfig = {
ExecStart = lib.escapeShellArgs ([
(lib.getExe cfg.package)
"--slice-us" (toString cfg.sliceUs)
"--starve-ms" (toString cfg.starveMs)
"--timeout-ms" (toString cfg.timeoutMs)
"--report-sec" (toString cfg.reportSec)
] ++ cfg.extraArgs);
Type = "simple";
Restart = "on-failure";
RestartSec = 5;
KillSignal = "SIGTERM";
TimeoutStopSec = 15;
# Losing the scheduler to the OOM killer would be a strange
# way to lose the scheduler.
OOMScoreAdjust = -500;
};
};
systemd.services.scx-soryu-notify = lib.mkIf cfg.notify.enable {
description = "Report that scx_soryu gave up";
serviceConfig = {
Type = "oneshot";
ExecStart = pkgs.writeShellScript "scx-soryu-notify" ''
set -u
echo "soryu panic!; the system is running default scheduler" >&2
echo "journalctl -u scx-soryu -n 200 has the exit reason" >&2
uid=$(${pkgs.coreutils}/bin/id -u ${cfg.notify.user}) || exit 0
bus=/run/user/$uid/bus
[ -S "$bus" ] || exit 0
${pkgs.util-linux}/bin/runuser -u ${cfg.notify.user} -- ${pkgs.coreutils}/bin/env "DBUS_SESSION_BUS_ADDRESS=unix:path=$bus" ${pkgs.libnotify}/bin/notify-send --urgency=critical --icon=dialog-error "scx_soryu durdu" "Sistem EEVDF'e döndü. journalctl -u scx-soryu" || true
'';
};
};
};
};
nixosConfigurations.testvm = nixpkgs.lib.nixosSystem {
inherit system;
modules = [
"${nixpkgs}/nixos/modules/virtualisation/qemu-vm.nix"
vmBase
];
};
nixosConfigurations.servicevm = nixpkgs.lib.nixosSystem {
inherit system;
modules = [
"${nixpkgs}/nixos/modules/virtualisation/qemu-vm.nix"
vmBase
self.nixosModules.default
{
services.scx-soryu = {
enable = true;
reportSec = 5;
# No desktop session in here, so nothing to notify.
notify.enable = false;
};
}
];
};
};
}
src/bpf/main.bpf.c
+751
View File
@@ -0,0 +1,751 @@
/* SPDX-License-Identifier: GPL-2.0 */
/*
* scx_soryu — BMQ policy on sched_ext.
*/
#include <scx/common.bpf.h>
char _license[] SEC("license") = "GPL";
UEI_DEFINE(uei);
/*
* BMQ compresses (nice + boost) by two, so adjacent nice levels share a queue.
* That halves nice resolution and in exchange gives boost twice the leverage
* per step. Keeping the compression is the point of porting BMQ rather than
* inventing a lookalike.
*
* BMQ: 25 + (prio + boost - 128) / 2, where prio = 160 + nice
* 0-based: (32 + nice + boost) / 2
* mainline: (static_prio - 88 + boost) / 2, nice = static_prio - 120
*/
#define BOOST_RANGE 12 /* BMQ MAX_PRIORITY_ADJ */
#define NORMAL_LEVELS 32 /* (12 + 40 + 12) / 2 */
#define LEVELS (NORMAL_LEVELS + 1) /* last level is SCHED_IDLE */
#define IDLE_LEVEL (LEVELS - 1)
#define MAX_CPUS 1024
/*
* BMQ's RESCHED_NS (alt_core.c:94). A remainder below this is treated as spent:
* handing back a sliver just makes the task come straight back, so BMQ refills
* instead. Same threshold, same reason.
*/
#define RESCHED_NS (100 << 10) /* ~102 us */
/* uapi/linux/sched.h; vmlinux.h carries the enum but not these defines. */
#define SCHED_NORMAL 0
#define SCHED_BATCH 3
#define SCHED_IDLE 5
#define DSQ_BASE 0x736f7279UL /* "sory" */
#define dsq_id(lvl) (DSQ_BASE + (lvl))
const volatile u64 slice_ns = 4194304ULL; /* 4 << 20, BMQ sysctl_sched_base_slice */
const volatile u32 nr_cpu_ids = 1; /* filled in by the loader */
/*
* BMQ's sched_task_ttwu() boosts on wakeup in proportion to how long the task
* slept, which is what makes sleeping for 1us worth nothing and closes the
* sleep-gaming hole. Except the line reads
*
* s64 delta = this_rq()->clock_task > p->last_ran;
*
* -- a comparison, not a subtraction. delta is always 0 or 1, the shift makes
* it 0, and the whole function is dead code in every BMQ build shipped.
*
* Fixing it means running behaviour nobody has field-tested; leaving it means
* reproducing a typo. So it is a knob, measured both ways in stage 2.
*/
const volatile bool boost_on_wake = true;
/*
* Anti-starvation, and a deliberate deviation from BMQ.
*
* BMQ has none. bmq.h is forty lines: boost comes from sleeping, deboost from
* burning a slice, and nothing whatsoever happens to a task that is merely
* waiting. Strict priority, no aging, no deadline. On mainline that is
* survivable -- a nice-19 thread stuck behind a compile is just slow, and no
* one dies. Under sched_ext the watchdog calls it a fatal stall and ejects the
* whole scheduler, which is what happened on 1 Sep: helium:disk$2 (nice 19,
* level 25) waited 7.115s behind seven compilers at level 22 and took us down
* with it. See docs/stage6-stall.md.
*
* So: a level that has held tasks this long without being served gets a turn,
* once, ahead of better levels. One task per level per starve_ns, which works
* out to roughly 3% of a CPU for a starving band -- close to what nice 19 is
* worth under CFS anyway (weight 15/1024). Strict priority is preserved
* everywhere else.
*
* 0 disables it and restores BMQ's semantics exactly.
*/
const volatile u64 starve_ns = 128ULL << 20; /* ~134 ms */
u64 lvl_bitmap;
/* Level each CPU is currently running, LEVELS when idle. Used by stage 4. */
static u8 cpu_level[MAX_CPUS];
u64 max_wait_jiffies;
u64 lvl_max_wait[LEVELS];
u64 lvl_last_run[LEVELS];
u32 starve_cursor;
u32 probe_seq;
/*
* How many times each boost value was seen at ops.running. The single most
* useful stage 2 number: it says whether boost is actually moving, and in which
* direction the workload pushes it. Index is boost + BOOST_RANGE.
*/
u64 boost_hist[2 * BOOST_RANGE + 1];
#define LAT_CLASSES 3 /* 0 boosted, 1 neutral, 2 deboosted */
#define LAT_BUCKETS 32 /* log2(ns): bucket 10 is ~1us, 30 is ~1s */
u64 lat_hist[LAT_CLASSES * LAT_BUCKETS];
struct task_ctx
{
s32 boost; /* [-BOOST_RANGE, +BOOST_RANGE], negative = better */
u64 sleep_at; /* scx_bpf_now() when the task went to sleep */
u64 run_at; /* scx_bpf_now() when it last started running */
u64 wake_at; /* scx_bpf_now() at ops.runnable, for lat_hist */
u64 slice_given; /* what p->scx.slice held at ops.running */
u64 slice_left;
};
struct
{
__uint(type, BPF_MAP_TYPE_TASK_STORAGE);
__uint(map_flags, BPF_F_NO_PREALLOC);
__type(key, int);
__type(value, struct task_ctx);
} task_ctxs SEC(".maps");
enum soryu_stat_idx
{
SORYU_STAT_ENQ_SHARED,
SORYU_STAT_ENQ_LOCAL,
SORYU_STAT_ENQ_REENQ,
SORYU_STAT_ENQ_PINNED,
SORYU_STAT_DISPATCH,
SORYU_STAT_INSERT_FAIL,
SORYU_STAT_CPU_RELEASE,
SORYU_STAT_BOOST_SLEEP,
SORYU_STAT_BOOST_WAKE,
SORYU_STAT_DEBOOST,
SORYU_STAT_SLICE_CARRY,
SORYU_STAT_FORK_SPLIT,
SORYU_STAT_PREEMPT_KICK,
SORYU_STAT_STARVE_SERVE,
SORYU_STAT_PREEMPT_NONE,
SORYU_STAT_PREEMPT_IDLE,
SORYU_NR_STATS,
};
struct
{
__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
__uint(key_size, sizeof(u32));
__uint(value_size, sizeof(u64));
__uint(max_entries, SORYU_NR_STATS);
} stats SEC(".maps");
static void stat_inc(enum soryu_stat_idx idx)
{
u32 key = idx;
u64 *cnt = bpf_map_lookup_elem(&stats, &key);
if (cnt)
(*cnt)++;
}
static struct task_ctx *lookup_task_ctx(struct task_struct *p)
{
return bpf_task_storage_get(&task_ctxs, p, 0, 0);
}
/*
* BMQ boost_task(): the floor depends on policy. SCHED_NORMAL may go all the
* way to -12; SCHED_BATCH may only climb back to 0, so a batch task can recover
* from a deboost but never outrank a neutral one; anything else (SCHED_IDLE)
* never boosts at all.
*/
static void boost_task(struct task_struct *p, struct task_ctx *tctx, u32 n)
{
s32 limit, want;
switch (p->policy)
{
case SCHED_NORMAL:
limit = -BOOST_RANGE;
break;
case SCHED_BATCH:
limit = 0;
break;
default:
return;
}
want = (s32)tctx->boost - (s32)n;
tctx->boost = want < limit ? limit : want;
}
/* BMQ deboost_task(): NORMAL and BATCH both decay, SCHED_IDLE does not. */
static void deboost_task(struct task_struct *p, struct task_ctx *tctx)
{
if ((p->policy == SCHED_NORMAL || p->policy == SCHED_BATCH) &&
tctx->boost < BOOST_RANGE)
tctx->boost++;
}
static u32 task_level(struct task_struct *p, struct task_ctx *tctx)
{
s32 v;
if (p->policy == SCHED_IDLE)
return IDLE_LEVEL;
/*
* static_prio in [100, 139], boost in [-12, 12], so v lands in [0, 63]
* and never goes negative — no clamp needed, and >> 1 == / 2.
*
* static_prio rather than p->scx.weight on purpose: weight is a
* logarithmic thing derived from nice, and the level math wants nice
* itself. ops.set_weight is deliberately not implemented.
*/
v = (s32)p->static_prio - 88 + (tctx ? tctx->boost : 0);
return (u32)v >> 1;
}
static u64 task_slice(struct task_ctx *tctx)
{
if (tctx && tctx->slice_left >= RESCHED_NS)
return tctx->slice_left;
return slice_ns;
}
/*
* Kick the CPU running the worst thing this task could displace.
*
* BMQ does this on every wakeup: if the woken task outranks what some CPU is
* running, send a resched IPI. With a shared queue there is no single "target
* CPU", so we track what level each CPU is running (cpu_level[]) and search.
*
* Two deliberate differences from LFBMQ (alt_core.c:4225):
*
* - It kicks *every* eligible CPU; we kick only the worst one. Broadcasting
* costs an IPI per CPU per wakeup, and stage 1 measured the shared queue
* taking 82% of wakeups under game load -- that is a lot of wakeups to
* multiply. One kick is enough: the preempted CPU re-enters dispatch and
* pulls the lowest level from the bitmap anyway.
*
* - It respects p->cpus_ptr and so do we. The spec originally dropped this;
* kicking a CPU the task cannot run on wastes the IPI and hands the slot to
* somebody else.
*/
static void try_preempt(struct task_struct *p, u32 lvl, u64 enq_flags)
{
const struct cpumask *allowed = p->cpus_ptr;
s32 worst_cpu = -1;
u32 worst = lvl + 1; /* never IPI an equal or better level */
u32 nr = scx_bpf_nr_cpu_ids();
u32 cpu;
if (!allowed)
return;
/*
* Wakeups only. BMQ checks for preemption in wakeup_srq_task()
* (alt_core.c:4211) -- on the wakeup path, not on every enqueue.
*
* Running it on every enqueue closes a loop: the task we preempt is
* re-enqueued, that re-enqueue runs the check again, which preempts
* somebody else, and so on. Measured at 3.2M kicks in 32s (~100k
* IPIs/sec) on four spinners plus one 100Hz waker -- three orders of
* magnitude more than the wakeups that caused it.
*/
if (!(enq_flags & SCX_ENQ_WAKEUP))
return;
bpf_for(cpu, 0, nr)
{
u32 cl;
if (cpu >= MAX_CPUS)
break;
if (!bpf_cpumask_test_cpu(cpu, allowed))
continue;
cl = cpu_level[cpu];
if (cl > worst)
{
worst = cl;
worst_cpu = cpu;
}
}
if (worst_cpu >= 0)
{
scx_bpf_kick_cpu(worst_cpu, SCX_KICK_PREEMPT);
/*
* Two very different events were hiding behind one counter.
*
* cpu_level is set to LEVELS when a CPU stops running an scx
* task, so an idle CPU always looks "worse" than any level and
* always wins the search. Most of what preempt_kick counted was
* therefore *waking an idle CPU*, not preempting anybody --
* measured in the VM at 11:1 against real preemption with eight
* sleepers. Necessary work, but it says nothing about
* preemption pressure, and the two must not share a number.
*/
if (worst >= LEVELS)
stat_inc(SORYU_STAT_PREEMPT_IDLE);
else
stat_inc(SORYU_STAT_PREEMPT_KICK);
return;
}
/*
* Nobody worse to kick, so this wakeup waits for a slice to end on its
* own -- up to slice_ns.
*
* This is the measurement behind the slice question. When the boost
* range saturates, every task sits at the same level, `worst` never
* exceeds `lvl`, and preemption switches itself off: the scheduler
* silently degrades to FIFO-with-a-4ms-quantum exactly in the workload
* (gaming, 97% at the floor) where responsiveness is the whole point.
* If this counter dwarfs preempt_kick there, the tail is a slice-length
* problem and no amount of boost tuning will touch it.
*/
stat_inc(SORYU_STAT_PREEMPT_NONE);
}
s32 BPF_STRUCT_OPS(soryu_select_cpu, struct task_struct *p, s32 prev_cpu,
u64 wake_flags)
{
struct task_ctx *tctx;
bool is_idle = false;
s32 cpu;
cpu = scx_bpf_select_cpu_dfl(p, prev_cpu, wake_flags, &is_idle);
if (!is_idle)
return cpu;
/*
* Hot path: an idle CPU was found, so skip the shared queue entirely.
* ops.enqueue is not called for tasks inserted from here.
*/
tctx = lookup_task_ctx(p);
if (scx_bpf_dsq_insert(p, SCX_DSQ_LOCAL, task_slice(tctx), 0))
stat_inc(SORYU_STAT_ENQ_LOCAL);
else
stat_inc(SORYU_STAT_INSERT_FAIL);
return cpu;
}
void BPF_STRUCT_OPS(soryu_enqueue, struct task_struct *p, u64 enq_flags)
{
struct task_ctx *tctx = lookup_task_ctx(p);
u32 lvl = task_level(p, tctx);
/*
* Tasks bounced back by ops.cpu_release did not stop through any fault
* of their own; stage 2 must not read this as a scheduling signal.
*/
if (enq_flags & SCX_ENQ_REENQ)
stat_inc(SORYU_STAT_ENQ_REENQ);
/*
* Single-CPU tasks gain nothing from a shared queue, and they make every
* other CPU pay: consume_dispatch_q() walks the whole DSQ list under
* dsq->lock skipping tasks it cannot migrate, so a pile of pinned tasks
* turns each remote dispatch into a futile scan. Measured at 64 pinned
* tasks: 6x slower than EEVDF before this, and it is exactly the
* live-lock the kernel comments about. Per-CPU kthreads are in this
* class too, so it is not an exotic case.
*
* Restricted-but-multi-CPU masks still go to the shared queue; they are
* rarer and the fix for them is per-CPU DSQs, which is a bigger change.
*/
if (p->nr_cpus_allowed == 1)
{
s32 cpu = scx_bpf_task_cpu(p);
if (scx_bpf_dsq_insert(p, SCX_DSQ_LOCAL_ON | cpu,
task_slice(tctx), enq_flags))
{
stat_inc(SORYU_STAT_ENQ_PINNED);
/* Only one CPU can ever run it, so the search above is
* pointless -- just check that one. Wakeups only, same
* reason as try_preempt(). */
if ((enq_flags & SCX_ENQ_WAKEUP) &&
cpu >= 0 && cpu < MAX_CPUS &&
cpu_level[cpu] > lvl)
{
scx_bpf_kick_cpu(cpu, SCX_KICK_PREEMPT);
stat_inc(SORYU_STAT_PREEMPT_KICK);
}
}
else
{
stat_inc(SORYU_STAT_INSERT_FAIL);
}
return;
}
if (!scx_bpf_dsq_insert(p, dsq_id(lvl), task_slice(tctx), enq_flags))
{
stat_inc(SORYU_STAT_INSERT_FAIL);
return;
}
stat_inc(SORYU_STAT_ENQ_SHARED);
/* Insert first, then advertise: a bit set late is a missed dispatch
* round, a bit set early is a dispatch that finds nothing. */
__sync_fetch_and_or(&lvl_bitmap, 1ULL << lvl);
try_preempt(p, lvl, enq_flags);
}
static void sample_head_wait(u32 lvl)
{
struct task_struct *p;
u64 now = bpf_jiffies64();
if (lvl >= LEVELS)
return;
bpf_for_each(scx_dsq, p, dsq_id(lvl), 0)
{
u64 runnable_at = p->scx.runnable_at;
if (runnable_at)
{
u64 waited = time_delta(now, runnable_at);
if (waited > READ_ONCE(max_wait_jiffies))
WRITE_ONCE(max_wait_jiffies, waited);
if (waited > READ_ONCE(lvl_max_wait[lvl]))
WRITE_ONCE(lvl_max_wait[lvl], waited);
}
break;
}
}
void BPF_STRUCT_OPS(soryu_dispatch, s32 cpu, struct task_struct *prev)
{
u64 mask = READ_ONCE(lvl_bitmap);
u64 now = scx_bpf_now();
if (starve_ns)
{
u32 probe = READ_ONCE(starve_cursor);
probe = probe < LEVELS - 1 ? probe + 1 : 0;
WRITE_ONCE(starve_cursor, probe);
/*
* Telemetry, once every 1024 dispatches -- tens of times a
* second, which is three orders of magnitude finer than a
* multi-second watchdog needs, and cheap enough to ignore.
* (1024 mod 33 = 1, so consecutive samples walk every level.)
*/
if (!(probe_seq++ & 1023) && (mask & (1ULL << probe)))
sample_head_wait(probe);
if ((mask & (1ULL << probe)) &&
now - READ_ONCE(lvl_last_run[probe]) > starve_ns)
{
if (scx_bpf_dsq_move_to_local(dsq_id(probe), 0))
{
WRITE_ONCE(lvl_last_run[probe], now);
stat_inc(SORYU_STAT_DISPATCH);
stat_inc(SORYU_STAT_STARVE_SERVE);
return;
}
/*
* Empty, or nothing here can run on this CPU. Either
* way it is not starving -- it was offered a turn and
* had nothing to take. Stamping it keeps a level that
* has been idle for hours (the bitmap is set-only, so
* its bit stays lit forever) from claiming a free pass
* on every probe.
*/
WRITE_ONCE(lvl_last_run[probe], now);
}
}
bpf_repeat(LEVELS)
{
u32 lvl;
u64 bit;
if (!mask)
break;
lvl = __builtin_ctzll(mask);
bit = 1ULL << lvl;
if (scx_bpf_dsq_move_to_local(dsq_id(lvl), 0))
{
WRITE_ONCE(lvl_last_run[lvl], now);
stat_inc(SORYU_STAT_DISPATCH);
return;
}
mask &= ~bit;
}
}
void BPF_STRUCT_OPS(soryu_runnable, struct task_struct *p, u64 enq_flags)
{
struct task_ctx *tctx;
u64 now, slept, steps;
if (enq_flags & SCX_ENQ_REENQ)
return;
tctx = lookup_task_ctx(p);
if (!tctx)
return;
now = scx_bpf_now();
tctx->wake_at = now;
if (!boost_on_wake || !tctx->sleep_at || !slice_ns)
return;
slept = now > tctx->sleep_at ? now - tctx->sleep_at : 0;
tctx->sleep_at = 0;
steps = slept / slice_ns;
if (!steps)
return;
if (steps > 2 * BOOST_RANGE)
steps = 2 * BOOST_RANGE;
boost_task(p, tctx, (u32)steps);
stat_inc(SORYU_STAT_BOOST_WAKE);
}
void BPF_STRUCT_OPS(soryu_running, struct task_struct *p)
{
struct task_ctx *tctx = lookup_task_ctx(p);
s32 cpu = scx_bpf_task_cpu(p);
u64 runnable_at, waited;
if (cpu >= 0 && cpu < MAX_CPUS)
cpu_level[cpu] = task_level(p, tctx);
if (tctx)
{
u32 idx = (u32)(tctx->boost + BOOST_RANGE);
if (idx < 2 * BOOST_RANGE + 1)
boost_hist[idx]++;
tctx->run_at = scx_bpf_now();
tctx->slice_given = p->scx.slice;
if (tctx->wake_at)
{
u64 lat = tctx->run_at > tctx->wake_at
? tctx->run_at - tctx->wake_at
: 0;
u32 cls = tctx->boost < 0 ? 0 : (tctx->boost ? 2 : 1);
u32 b = lat ? 63 - __builtin_clzll(lat) : 0;
u32 slot;
if (b >= LAT_BUCKETS)
b = LAT_BUCKETS - 1;
slot = cls * LAT_BUCKETS + b;
if (slot < LAT_CLASSES * LAT_BUCKETS)
lat_hist[slot]++;
tctx->wake_at = 0;
}
}
runnable_at = p->scx.runnable_at;
if (runnable_at)
{
u32 lvl = task_level(p, tctx);
waited = time_delta(bpf_jiffies64(), runnable_at);
if (waited > READ_ONCE(max_wait_jiffies))
WRITE_ONCE(max_wait_jiffies, waited);
if (lvl < LEVELS && waited > READ_ONCE(lvl_max_wait[lvl]))
WRITE_ONCE(lvl_max_wait[lvl], waited);
}
}
void BPF_STRUCT_OPS(soryu_stopping, struct task_struct *p, bool runnable)
{
struct task_ctx *tctx = lookup_task_ctx(p);
s32 cpu = scx_bpf_task_cpu(p);
u64 now, ran, left;
if (cpu >= 0 && cpu < MAX_CPUS)
cpu_level[cpu] = LEVELS;
if (!tctx)
return;
if (!runnable)
{
boost_task(p, tctx, 1);
tctx->sleep_at = scx_bpf_now();
tctx->slice_left = 0; /* sleeping earns a fresh slice */
stat_inc(SORYU_STAT_BOOST_SLEEP);
return;
}
now = scx_bpf_now();
ran = now > tctx->run_at ? now - tctx->run_at : 0;
left = tctx->slice_given > ran ? tctx->slice_given - ran : 0;
if (left < RESCHED_NS)
{
deboost_task(p, tctx);
tctx->slice_left = 0;
stat_inc(SORYU_STAT_DEBOOST);
}
else
{
tctx->slice_left = left;
stat_inc(SORYU_STAT_SLICE_CARRY);
}
}
void BPF_STRUCT_OPS(soryu_cpu_release, s32 cpu, struct scx_cpu_release_args *args)
{
scx_bpf_reenqueue_local();
if (cpu >= 0 && cpu < MAX_CPUS)
cpu_level[cpu] = LEVELS;
stat_inc(SORYU_STAT_CPU_RELEASE);
}
s32 BPF_STRUCT_OPS(soryu_init_task, struct task_struct *p,
struct scx_init_task_args *args)
{
struct task_ctx *tctx;
tctx = bpf_task_storage_get(&task_ctxs, p, 0,
BPF_LOCAL_STORAGE_GET_F_CREATE);
if (!tctx)
return -ENOMEM;
tctx->boost = 0;
tctx->sleep_at = 0;
tctx->run_at = 0;
tctx->wake_at = 0;
tctx->slice_given = 0;
tctx->slice_left = 0;
/*
* BMQ (alt_core.c:2874, sched_cgroup_fork):
*
* rq->curr->time_slice /= 2;
* p->time_slice = rq->curr->time_slice;
*
* with the comment "the total amount of pending timeslices in the
* system doesn't change". Without it every fork mints a fresh full
* slice out of nothing, so anything that forks in a loop -- a build,
* most obviously -- gains CPU share against everything else simply by
* forking. Splitting closes that.
*
*/
if (args->fork)
{
struct task_struct *parent = bpf_get_current_task_btf();
struct task_ctx *pctx;
u64 half;
if (parent)
{
half = parent->scx.slice / 2;
scx_bpf_task_set_slice(parent, half);
pctx = bpf_task_storage_get(&task_ctxs, parent, 0, 0);
if (pctx)
pctx->slice_left = half;
tctx->slice_left = half >= RESCHED_NS ? half : 0;
stat_inc(SORYU_STAT_FORK_SPLIT);
}
}
return 0;
}
void BPF_STRUCT_OPS(soryu_dump, struct scx_dump_ctx *dctx)
{
u64 mask = READ_ONCE(lvl_bitmap);
u32 lvl;
scx_bpf_dump_header();
scx_bpf_dump("soryu: bitmap=0x%llx max_wait=%llu jiffies\n",
mask, READ_ONCE(max_wait_jiffies));
bpf_for(lvl, 0, LEVELS)
{
s32 nr = scx_bpf_dsq_nr_queued(dsq_id(lvl));
/* [BIT CLEAR!] is an invariant violation: a level holding tasks
* whose bit is unset can never be dispatched. Must never fire
* now that bits are set-only. */
if (nr > 0)
scx_bpf_dump(" lvl %2u: %d queued%s\n", lvl, nr,
(mask & (1ULL << lvl)) ? "" : " [BIT CLEAR!]");
}
}
s32 BPF_STRUCT_OPS_SLEEPABLE(soryu_init)
{
u32 lvl;
s32 ret;
bpf_for(lvl, 0, LEVELS)
{
ret = scx_bpf_create_dsq(dsq_id(lvl), -1);
if (ret)
return ret;
}
bpf_for(lvl, 0, MAX_CPUS)
cpu_level[lvl] = LEVELS;
return 0;
}
void BPF_STRUCT_OPS(soryu_exit, struct scx_exit_info *ei)
{
UEI_RECORD(uei, ei);
}
SCX_OPS_DEFINE(soryu_ops,
.select_cpu = (void *)soryu_select_cpu,
.enqueue = (void *)soryu_enqueue,
.dispatch = (void *)soryu_dispatch,
.runnable = (void *)soryu_runnable,
.running = (void *)soryu_running,
.stopping = (void *)soryu_stopping,
.cpu_release = (void *)soryu_cpu_release,
.init_task = (void *)soryu_init_task,
.dump = (void *)soryu_dump,
.init = (void *)soryu_init,
.exit = (void *)soryu_exit,
.flags = SCX_OPS_KEEP_BUILTIN_IDLE,
.timeout_ms = 5000,
.name = "soryu");
src/bpf_skel.rs
+7
View File
@@ -0,0 +1,7 @@
// SPDX-License-Identifier: GPL-2.0
//
// libbpf-cargo generates the skeleton into OUT_DIR; this module just pulls it
// in. Kept separate so the generated code's lint profile stays isolated.
#![allow(dead_code)]
#![allow(clippy::all)]
include!(concat!(env!("OUT_DIR"), "/bpf_skel.rs"));
src/main.rs
+502
View File
@@ -0,0 +1,502 @@
// SPDX-License-Identifier: GPL-2.0
//
// scx_soryu loader.
//
// runs in the foreground and holds no state of its own: dropping
// the struct_ops link is what unloads the scheduler, so Ctrl-C is the emergency
// exit during development.
mod bpf_skel;
mod stats;
use bpf_skel::*;
use stats::Metrics;
use std::mem::MaybeUninit;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use clap::Parser;
use crossbeam::channel::RecvTimeoutError;
use scx_stats::prelude::*;
use libbpf_rs::OpenObject;
use libbpf_rs::MapCore;
use log::error;
use log::info;
use log::warn;
use scx_utils::scx_ops_attach;
use scx_utils::scx_ops_load;
use scx_utils::scx_ops_open;
use scx_utils::uei_exited;
use scx_utils::uei_report;
use scx_utils::UserExitInfo;
const SCHEDULER_NAME: &str = "scx_soryu";
/// mirrors `enum stat_idx` in main.bpf.c.
const STAT_NAMES: &[&str] = &[
"enq_shared",
"enq_local",
"enq_reenq",
"enq_pinned",
"dispatch",
"insert_fail",
"cpu_release",
"boost_sleep",
"boost_wake",
"deboost",
"slice_carry",
"fork_split",
"preempt_kick",
"starve_serve",
"preempt_none",
"preempt_idle",
];
/// mirrors BOOST_RANGE in main.bpf.c.
const BOOST_RANGE: i32 = 12;
/// mirrors LAT_CLASSES / LAT_BUCKETS in main.bpf.c.
const LAT_CLASSES: usize = 3;
const LAT_BUCKETS: usize = 32;
const LAT_CLASS_NAMES: [&str; LAT_CLASSES] = ["boosted", "neutral", "deboosted"];
#[derive(Parser, Debug, Clone)]
#[command(about = "BMQ policy port for sched_ext", version)]
struct Opts {
#[arg(long, default_value_t = 5000, env = "SORYU_TIMEOUT_MS")]
timeout_ms: u32,
#[arg(long, default_value_t = true, env = "SORYU_BOOST_ON_WAKE",
action = clap::ArgAction::Set)]
boost_on_wake: bool,
#[arg(long, default_value_t = 1000, env = "SORYU_SLICE_US")]
slice_us: u64,
#[arg(long, default_value_t = 128, env = "SORYU_STARVE_MS")]
starve_ms: u64,
#[arg(long, default_value_t = 1, env = "SORYU_REPORT_SEC")]
report_sec: u64,
#[arg(long, env = "SORYU_LOG")]
log: Option<String>,
/// Print libbpf's output, including the verifier log on a load failure.
#[arg(long, default_value_t = false, env = "SORYU_DEBUG")]
debug: bool,
/// Bytes of scheduler state captured when the kernel ejects us.
#[arg(long, default_value_t = 65536)]
exit_dump_len: u32,
/// Attach to a running scx_soryu and print its telemetry every N seconds.
/// Starts no scheduler of its own.
#[arg(long)]
monitor: Option<f64>,
}
fn percentiles(buckets: &[u64]) -> Option<(u64, u64, u64)> {
let total: u64 = buckets.iter().sum();
if total == 0 {
return None;
}
let at = |frac: f64| -> u64 {
let target = (total as f64 * frac) as u64;
let mut seen = 0u64;
for (i, &c) in buckets.iter().enumerate() {
seen += c;
if seen >= target {
return 1u64 << i;
}
}
1u64 << (buckets.len() - 1)
};
Some((at(0.50), at(0.99), at(0.999)))
}
fn fmt_ns(ns: u64) -> String {
if ns >= 1_000_000_000 {
format!("{:.1}s", ns as f64 / 1e9)
} else if ns >= 1_000_000 {
format!("{}ms", ns / 1_000_000)
} else if ns >= 1_000 {
format!("{}us", ns / 1_000)
} else {
format!("{ns}ns")
}
}
struct Scheduler<'a> {
opts: Opts,
skel: BpfSkel<'a>,
struct_ops: Option<libbpf_rs::Link>,
stats_server: StatsServer<(), Metrics>,
prev_waits: Vec<u64>,
}
impl<'a> Scheduler<'a> {
fn init(opts: Opts, open_object: &'a mut MaybeUninit<OpenObject>) -> Result<Self> {
let mut skel_builder = BpfSkelBuilder::default();
// SORYU_DEBUG=1 turns on libbpf's output, which is the only way to see
// the verifier's rejection log when a program fails to load.
skel_builder.obj_builder.debug(opts.debug);
let open_opts = scx_utils::libbpf_clap_opts::LibbpfOpts::default().into_bpf_open_opts();
let mut skel = scx_ops_open!(skel_builder, open_object, soryu_ops, open_opts)?;
// Dump the scheduler's own state when we get ejected; ops.dump fills it.
skel.struct_ops.soryu_ops_mut().exit_dump_len = opts.exit_dump_len;
skel.struct_ops.soryu_ops_mut().timeout_ms = opts.timeout_ms;
let rodata = skel
.maps
.rodata_data
.as_mut()
.context("no rodata section in skeleton")?;
rodata.nr_cpu_ids = *scx_utils::NR_CPU_IDS as u32;
// SORYU_BOOST_ON_WAKE=0 reproduces BMQ as shipped (its wake boost is
// dead code); the default reproduces what BMQ meant to do.
rodata.boost_on_wake = opts.boost_on_wake;
rodata.starve_ns = opts.starve_ms * 1_000_000;
rodata.slice_ns = opts.slice_us * 1_000;
let mut skel = scx_ops_load!(skel, soryu_ops, uei)?;
let struct_ops = Some(scx_ops_attach!(skel, soryu_ops)?);
let rd = skel.maps.rodata_data.as_ref().unwrap();
info!(
"{} attached: {} CPUs, slice {} us, boost_on_wake={}, \
starve {} ms, timeout {} ms",
SCHEDULER_NAME,
*scx_utils::NR_CPU_IDS,
rd.slice_ns / 1000,
rd.boost_on_wake,
rd.starve_ns / 1_000_000,
opts.timeout_ms
);
let stats_server = StatsServer::new(stats::server_data()).launch()?;
Ok(Self {
opts,
skel,
struct_ops,
stats_server,
prev_waits: Vec::new(),
})
}
fn exited(&mut self) -> bool {
uei_exited!(&self.skel, uei)
}
/// Per-CPU counters summed across CPUs.
fn read_stats(&self) -> Result<Vec<u64>> {
let nr_cpus = *scx_utils::NR_CPU_IDS;
let mut out = vec![0u64; STAT_NAMES.len()];
for (idx, slot) in out.iter_mut().enumerate() {
let key = (idx as u32).to_ne_bytes();
let per_cpu = self
.skel
.maps
.stats
.lookup_percpu(&key, libbpf_rs::MapFlags::ANY)?
.unwrap_or_default();
*slot = per_cpu
.iter()
.take(nr_cpus)
.map(|v| u64::from_ne_bytes(v[..8].try_into().unwrap_or([0; 8])))
.sum();
}
Ok(out)
}
/// One snapshot for the stats server.
fn metrics(&self) -> Metrics {
let c = self.read_stats().unwrap_or_default();
let g = |i: usize| c.get(i).copied().unwrap_or(0);
let mut m = Metrics {
enq_shared: g(0),
enq_local: g(1),
enq_reenq: g(2),
enq_pinned: g(3),
insert_fail: g(5),
cpu_release: g(6),
boost_sleep: g(7),
boost_wake: g(8),
deboost: g(9),
slice_carry: g(10),
fork_split: g(11),
preempt_kick: g(12),
starve_serve: g(13),
preempt_none: g(14),
preempt_idle: g(15),
..Default::default()
};
if let Some(bss) = self.skel.maps.bss_data.as_ref() {
m.max_wait_j = bss.max_wait_jiffies;
m.lvl_bitmap = bss.lvl_bitmap;
let hist = &bss.boost_hist;
let total: u64 = hist.iter().sum();
if total > 0 {
let weighted: i64 = hist
.iter()
.enumerate()
.map(|(i, &n)| (i as i64 - BOOST_RANGE as i64) * n as i64)
.sum();
m.boost_mean_x100 = weighted * 100 / total as i64;
m.boost_floor_pct = hist[0] * 100 / total;
m.boost_ceil_pct = hist[hist.len() - 1] * 100 / total;
}
// Class 0 is "boosted" -- the interactive tail, and the only class
// with enough samples to mean anything (hogs barely ever wake).
let boosted = &bss.lat_hist[0..LAT_BUCKETS];
m.lat_boosted_n = boosted.iter().sum();
if let Some((p50, p99, p999)) = percentiles(boosted) {
m.lat_boosted_p50 = p50;
m.lat_boosted_p99 = p99;
m.lat_boosted_p999 = p999;
}
}
m
}
fn report(&mut self) {
let stats = match self.read_stats() {
Ok(s) => s,
Err(e) => {
warn!("stat read failed: {e}");
return;
}
};
let bss = self.skel.maps.bss_data.as_ref();
let (bitmap, max_wait) = bss
.map(|b| (b.lvl_bitmap, b.max_wait_jiffies))
.unwrap_or((0, 0));
let raw: Vec<u64> = self
.skel
.maps
.bss_data
.as_ref()
.map(|b| b.lvl_max_wait.to_vec())
.unwrap_or_default();
let waits: Vec<String> = raw
.iter()
.enumerate()
.filter(|(_, &w)| w > 0)
.map(|(lvl, &w)| {
let rising = self.prev_waits.get(lvl).map(|&p| w > p).unwrap_or(false);
let mark = if rising { "*" } else { "" };
match lvl {
32 => format!("idle:{w}j{mark}"),
10 => format!("l10*n0boosted:{w}j{mark}"),
16 => format!("l16*n0neutral:{w}j{mark}"),
22 => format!("l22*n0hog:{w}j{mark}"),
_ => format!("l{lvl}:{w}j{mark}"),
}
})
.collect();
if !raw.is_empty() {
self.prev_waits = raw;
}
let line = STAT_NAMES
.iter()
.zip(stats.iter())
.map(|(n, v)| format!("{n}={v}"))
.collect::<Vec<_>>()
.join(" ");
info!("{line} bitmap=0x{bitmap:x} max_wait={max_wait}j");
if !waits.is_empty() {
info!(" per-level worst wait: {}", waits.join(" "));
}
if let Some(bss) = self.skel.maps.bss_data.as_ref() {
let hist = &bss.boost_hist;
let total: u64 = hist.iter().sum();
if total > 0 {
let occupied: Vec<String> = hist
.iter()
.enumerate()
.filter(|(_, &c)| c > 0)
.map(|(i, &c)| {
let boost = i as i32 - BOOST_RANGE;
format!("{boost:+}:{:.0}%", 100.0 * c as f64 / total as f64)
})
.collect();
let weighted: i64 = hist
.iter()
.enumerate()
.map(|(i, &c)| (i as i64 - BOOST_RANGE as i64) * c as i64)
.sum();
let mean = weighted as f64 / total as f64;
info!(" boost mean={mean:+.2} dist: {}", occupied.join(" "));
}
let lines: Vec<String> = (0..LAT_CLASSES)
.filter_map(|c| {
let slice = &bss.lat_hist[c * LAT_BUCKETS..(c + 1) * LAT_BUCKETS];
percentiles(slice).map(|(p50, p99, p999)| {
format!(
"{}: p50={} p99={} p99.9={} n={}",
LAT_CLASS_NAMES[c],
fmt_ns(p50),
fmt_ns(p99),
fmt_ns(p999),
slice.iter().sum::<u64>()
)
})
})
.collect();
for l in lines {
info!(" wake->run {l}");
}
let boosted = &bss.lat_hist[0..LAT_BUCKETS];
let n: u64 = boosted.iter().sum();
if n > 0 {
let above = |bucket: usize| -> f64 {
100.0 * boosted[bucket..].iter().sum::<u64>() as f64 / n as f64
};
info!(
" wake->run boosted tail: >{} {:.3}% >{} {:.3}% >{} {:.4}% >{} {:.4}%",
fmt_ns(1 << 20),
above(20),
fmt_ns(1 << 22),
above(22),
fmt_ns(1 << 24),
above(24),
fmt_ns(1 << 26),
above(26),
);
}
}
}
fn run(&mut self, shutdown: Arc<AtomicBool>) -> Result<UserExitInfo> {
let (res_ch, req_ch) = self.stats_server.channels();
let every = self.opts.report_sec.max(1);
let mut ticks = 0u64;
while !shutdown.load(Ordering::Relaxed) && !self.exited() {
match req_ch.recv_timeout(Duration::from_secs(1)) {
Ok(()) => {
let m = self.metrics();
res_ch.send(m)?;
}
Err(RecvTimeoutError::Timeout) => {
ticks += 1;
if ticks % every == 0 {
self.report();
}
}
Err(e) => Err(e)?,
}
}
// Dropping the link is what detaches the scheduler.
let _ = self.struct_ops.take();
uei_report!(&self.skel, uei)
}
}
impl Drop for Scheduler<'_> {
fn drop(&mut self) {
info!("{SCHEDULER_NAME} unloaded, tasks returned to default kernel scheduler");
}
}
fn main() -> Result<()> {
let opts = Opts::parse();
let cfg = simplelog::ConfigBuilder::new()
.set_time_offset_to_local()
.expect("failed to set local time offset")
.build();
// SORYU_LOG=<path> writes the same output to a file as well.
let mut sinks: Vec<Box<dyn simplelog::SharedLogger>> = vec![simplelog::TermLogger::new(
simplelog::LevelFilter::Info,
cfg.clone(),
simplelog::TerminalMode::Stderr,
simplelog::ColorChoice::Auto,
)];
if let Some(path) = opts.log.clone() {
let file = std::fs::File::create(&path)
.with_context(|| format!("cannot open SORYU_LOG file {path:?}"))?;
sinks.push(simplelog::WriteLogger::new(
simplelog::LevelFilter::Info,
cfg,
file,
));
}
simplelog::CombinedLogger::init(sinks)?;
let shutdown = Arc::new(AtomicBool::new(false));
let shutdown_handler = shutdown.clone();
ctrlc::set_handler(move || {
shutdown_handler.store(true, Ordering::Relaxed);
})?;
if let Some(intv) = opts.monitor {
return stats::monitor(Duration::from_secs_f64(intv), shutdown);
}
info!(
"pid {} — Ctrl-C to unload, or from another terminal: sudo kill -INT {}",
std::process::id(),
std::process::id()
);
let mut open_object = MaybeUninit::uninit();
loop {
if shutdown.load(Ordering::Relaxed) {
break;
}
let mut sched = Scheduler::init(opts.clone(), &mut open_object)?;
let uei = match sched.run(shutdown.clone()) {
Ok(uei) => uei,
Err(e) => {
error!("{SCHEDULER_NAME} was ejected by the kernel: {e:#}");
error!("all tasks are back on default kernel scheduler; the soryu is NOT running");
return Err(e);
}
};
if !uei.should_restart() {
break;
}
if shutdown.load(Ordering::Relaxed) {
break;
}
info!("hotplug detected, reloading");
}
info!("{SCHEDULER_NAME} exiting");
Ok(())
}
src/stats.rs
+197
View File
@@ -0,0 +1,197 @@
// SPDX-License-Identifier: GPL-2.0
//
// Live telemetry, readable from a second terminal with `scx_soryu --monitor`
// while the scheduler runs.
use std::io::Write;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use scx_stats::prelude::*;
use scx_stats_derive::stat_doc;
use scx_stats_derive::Stats;
use serde::Deserialize;
use serde::Serialize;
#[stat_doc]
#[derive(Clone, Debug, Default, Serialize, Deserialize, Stats)]
#[stat(top)]
pub struct Metrics {
#[stat(desc = "Wakeups that found an idle CPU and skipped the shared queue")]
pub enq_local: u64,
#[stat(desc = "Enqueues onto a shared level DSQ")]
pub enq_shared: u64,
#[stat(desc = "Enqueues of single-CPU tasks, sent straight to that CPU")]
pub enq_pinned: u64,
#[stat(desc = "Tasks bounced back by ops.cpu_release")]
pub enq_reenq: u64,
#[stat(desc = "DSQ inserts that failed")]
pub insert_fail: u64,
#[stat(desc = "CPUs taken by a higher sched class (PipeWire RT, i915)")]
pub cpu_release: u64,
#[stat(desc = "Boosts from going to sleep")]
pub boost_sleep: u64,
#[stat(desc = "Boosts from waking after sleeping at least one slice")]
pub boost_wake: u64,
#[stat(desc = "Deboosts from burning a whole slice")]
pub deboost: u64,
#[stat(desc = "Preemptions that banked an unused slice remainder")]
pub slice_carry: u64,
#[stat(desc = "Forks that split the parent's remaining slice")]
pub fork_split: u64,
#[stat(desc = "Preemption IPIs sent")]
pub preempt_kick: u64,
#[stat(desc = "Dispatches that jumped priority to unstarve a level")]
pub starve_serve: u64,
#[stat(desc = "Wakeups with nobody worse to preempt -- they wait out a slice")]
pub preempt_none: u64,
#[stat(desc = "Wakeups that kicked an idle CPU rather than preempting anyone")]
pub preempt_idle: u64,
#[stat(desc = "Mean boost, x100 (negative is more interactive)")]
pub boost_mean_x100: i64,
#[stat(desc = "Percent of scheduling events at the -12 boost floor")]
pub boost_floor_pct: u64,
#[stat(desc = "Percent at the +12 ceiling")]
pub boost_ceil_pct: u64,
#[stat(desc = "Interactive wakeup->run p50, ns")]
pub lat_boosted_p50: u64,
#[stat(desc = "Interactive wakeup->run p99, ns -- the number that matters")]
pub lat_boosted_p99: u64,
#[stat(desc = "Interactive wakeup->run p99.9, ns")]
pub lat_boosted_p999: u64,
#[stat(desc = "Interactive wakeup samples")]
pub lat_boosted_n: u64,
#[stat(desc = "Worst wait by any runnable task since load, jiffies")]
pub max_wait_j: u64,
#[stat(desc = "Levels currently holding tasks")]
pub lvl_bitmap: u64,
}
impl Metrics {
fn fmt_ns(ns: u64) -> String {
if ns >= 1_000_000_000 {
format!("{:.1}s", ns as f64 / 1e9)
} else if ns >= 1_000_000 {
format!("{}ms", ns / 1_000_000)
} else if ns >= 1_000 {
format!("{}us", ns / 1_000)
} else {
format!("{ns}ns")
}
}
pub fn format<W: Write>(&self, w: &mut W) -> Result<()> {
writeln!(
w,
"wake->run interactive p50 {:>7} p99 {:>7} p99.9 {:>7} (n={})",
Self::fmt_ns(self.lat_boosted_p50),
Self::fmt_ns(self.lat_boosted_p99),
Self::fmt_ns(self.lat_boosted_p999),
self.lat_boosted_n,
)?;
writeln!(
w,
"boost mean {:+.2} floor {}% ceiling {}% max_wait {}j bitmap 0x{:x}",
self.boost_mean_x100 as f64 / 100.0,
self.boost_floor_pct,
self.boost_ceil_pct,
self.max_wait_j,
self.lvl_bitmap,
)?;
writeln!(
w,
"enqueue local {} shared {} pinned {} reenq {} fail {}",
self.enq_local,
self.enq_shared,
self.enq_pinned,
self.enq_reenq,
self.insert_fail,
)?;
writeln!(
w,
"signal boost_sleep {} boost_wake {} deboost {} carry {} fork_split {}",
self.boost_sleep,
self.boost_wake,
self.deboost,
self.slice_carry,
self.fork_split,
)?;
writeln!(
w,
"preempt kick {} idle {} none {} cpu_release {} starve_serve {}",
self.preempt_kick,
self.preempt_idle,
self.preempt_none,
self.cpu_release,
self.starve_serve,
)?;
Ok(())
}
fn delta(&self, rhs: &Self) -> Self {
Self {
enq_local: self.enq_local - rhs.enq_local,
enq_shared: self.enq_shared - rhs.enq_shared,
enq_pinned: self.enq_pinned - rhs.enq_pinned,
enq_reenq: self.enq_reenq - rhs.enq_reenq,
insert_fail: self.insert_fail - rhs.insert_fail,
cpu_release: self.cpu_release - rhs.cpu_release,
boost_sleep: self.boost_sleep - rhs.boost_sleep,
boost_wake: self.boost_wake - rhs.boost_wake,
deboost: self.deboost - rhs.deboost,
slice_carry: self.slice_carry - rhs.slice_carry,
fork_split: self.fork_split - rhs.fork_split,
preempt_kick: self.preempt_kick - rhs.preempt_kick,
starve_serve: self.starve_serve - rhs.starve_serve,
preempt_none: self.preempt_none - rhs.preempt_none,
preempt_idle: self.preempt_idle - rhs.preempt_idle,
lat_boosted_n: self.lat_boosted_n - rhs.lat_boosted_n,
..self.clone()
}
}
}
pub fn server_data() -> StatsServerData<(), Metrics> {
let open: Box<dyn StatsOpener<(), Metrics>> = Box::new(move |(req_ch, res_ch)| {
req_ch.send(())?;
let mut prev = res_ch.recv()?;
let read: Box<dyn StatsReader<(), Metrics>> = Box::new(move |_args, (req_ch, res_ch)| {
req_ch.send(())?;
let cur = res_ch.recv()?;
let delta = cur.delta(&prev);
prev = cur;
delta.to_json()
});
Ok(read)
});
StatsServerData::new()
.add_meta(Metrics::meta())
.add_ops("top", StatsOps { open, close: None })
}
pub fn monitor(intv: Duration, shutdown: Arc<AtomicBool>) -> Result<()> {
scx_utils::monitor_stats::<Metrics>(
&[],
intv,
|| shutdown.load(Ordering::Relaxed),
|metrics| metrics.format(&mut std::io::stdout()),
)
.with_context(|| {
format!(
"cannot reach the {} stats socket \
(/var/run/scx/root/stats); is it running, and did you use sudo?",
crate::SCHEDULER_NAME
)
})
}