diff --git a/skills/kvb-org-queue/SKILL.md b/skills/kvb-org-queue/SKILL.md index bc3a3cf..6de86a8 100644 --- a/skills/kvb-org-queue/SKILL.md +++ b/skills/kvb-org-queue/SKILL.md @@ -32,18 +32,18 @@ Light runs (no Docker, about a minute) self-admit with ## Order and grant -Every 20 s each waiting hold evaluates the whole queue under one lock: +Every 20 s each waiting hold evaluates the queue under one lock: 1. Hotfix jobs first, then the register rank (`rasci.py show`), one place earlier per whole hour waited, then arrival. 2. In that order, a job is blocked at disk stage stop or critical, at warn when it needs disk, at memory stage stop, during a radio-silence mayday, - while its kind's slots are taken, or while its workdir runs another job. + while its kind's slots are taken, or while its workdir runs another job; + `exclusive_kinds` run only beside `light_kinds`. 3. A job whose disk need would cut below the stop threshold, or whose footprint would push the running ones past `budget_percent`, reserves it against the jobs behind it; a job alone always fits. -4. An unblocked job counts as granted for the jobs behind it: none takes its - slot. +4. An unblocked job counts as granted for the jobs behind it. 5. Every job, alone too, needs CPU, and RAM used plus what it and the running jobs may still take, below `load_limits`. @@ -56,12 +56,12 @@ kind, minutes, disk, RAM and CPU) to `queue-history.jsonl`; the From the last 30 days, once 5 runs exist: - per signature, the disk need is p90 disk taken plus 1 GB; -- per kind, the slot limit is how many p90 RAM and CPU shares fit between the - median load at the grant and `load_limits`, at least 1; -- per job, the footprint is its signature's p90 share, else its kind's. +- per kind, the slot limit is how many p90 RAM and median CPU shares fit + between the median load at the grant and `load_limits`, at least 1; +- per job, the footprint is its signature's share, else its kind's. Until then the seeds hold: `slot_need_gb[kind]` for disk, 1 slot per kind, -the whole budget as footprint: it runs alone. +the whole budget as footprint. The optional `slot_limits {kind: N}` only caps a limit from above. ## Inspect and cancel diff --git a/skills/kvb-org-queue/scripts/history.py b/skills/kvb-org-queue/scripts/history.py index 6326053..e179839 100644 --- a/skills/kvb-org-queue/scripts/history.py +++ b/skills/kvb-org-queue/scripts/history.py @@ -28,10 +28,11 @@ INTERPRETERS = re.compile(r"python[\d.]*|bash|sh|zsh|node|perl|ruby") MAKE_VALUE_OPTIONS = frozenset({"-C", "-f", "-I", "-o", "-W"}) RESOURCES = ("ram", "cpu") SIGNATURE_HEADER = ("Signature", "Kind", "Runs", "Median min", "p90 disk GB", "p90 RAM %", "Need GB") -KIND_HEADER = ("Kind", "Runs", "p90 RAM share %", "p90 CPU share %", "Slot limit", "Source") +KIND_HEADER = ("Kind", "Runs", "p90 RAM share %", "Median CPU share %", "Slot limit", "Source") def first_operand(tokens: Iterable[str]) -> list[str]: - return [token for token in tokens if not token.startswith("-")][:1] + operand = next((token for token in tokens if not token.startswith("-")), None) + return [] if operand is None or os.sep in operand else [operand] def make_target(tokens: Iterator[str]) -> list[str]: @@ -56,7 +57,8 @@ def signature(argv: list[str]) -> str: ` for an interpreter running a script (python, bash, ...) and ` -c` for inline code; else ` `. An operand is an argument not starting with `-`; a missing - target or operand is left out. + target or operand is left out, and so is an operand holding a path + separator. """ name = os.path.basename(argv[0]) tokens = iter(argv[1:]) @@ -156,7 +158,7 @@ def command_stats(rows: list[dict]) -> dict: disk = p90([r["disk_gb"] for r in rows]) stats.update( minutes=statistics.median(r["minutes"] for r in rows), disk_gb=disk, - ram_share=p90(shares(rows, "mem")), cpu_share=p90(shares(rows, "cpu")), + ram_share=p90(shares(rows, "mem")), cpu_share=statistics.median(shares(rows, "cpu")), need_gb=math.ceil(disk + DISK_MARGIN_GB), ) return stats @@ -170,7 +172,7 @@ def fit(ceiling: float, rows: list[dict], resource: str, share: float) -> int: def kind_stats(rows: list[dict], limits: dict) -> dict: stats = {"runs": len(rows), "ram_share": None, "cpu_share": None, "limit": None} if len(rows) >= MIN_RUNS: - ram, cpu = p90(shares(rows, "mem")), p90(shares(rows, "cpu")) + ram, cpu = p90(shares(rows, "mem")), statistics.median(shares(rows, "cpu")) stats.update(ram_share=ram, cpu_share=cpu, limit=max(1, min( fit(limits["mem_percent"], rows, "mem", ram), fit(limits["cpu_percent"], rows, "cpu", cpu), ))) @@ -190,9 +192,9 @@ def learn(rows: list[dict], now: float, cfg: dict) -> tuple[dict[str, dict], dic (signatures, kinds). signatures: {signature: {kinds, runs, minutes, disk_gb, ram_share, cpu_share, need_gb}}; kinds: {kind: {runs, ram_share, cpu_share, limit}}. With fewer than MIN_RUNS rows every value but kinds - and runs is None. minutes is the median, disk_gb, ram_share and - cpu_share the p90 of the disk taken and of max minus before in - percentage points; need_gb is ceil(disk_gb + DISK_MARGIN_GB); limit is + and runs is None. minutes is the median, disk_gb and ram_share the p90 + of the disk taken and of max minus before in percentage points, + cpu_share the median of max minus before; need_gb is ceil(disk_gb + DISK_MARGIN_GB); limit is max(1, min over RAM and CPU of floor((load limit - median before) / max(share, MIN_SHARE_PERCENT))). """ @@ -212,7 +214,7 @@ def footprint(kind: str, name: str, signatures: dict[str, dict], kinds: dict[str kinds: the kind map of learn(). Returns: - {ram, cpu}: the signature's p90 share once learned, else the kind's, at + {ram, cpu}: the signature's learned share once learned, else the kind's, at least MIN_SHARE_PERCENT. None while the kind is not learned yet. """ if kind not in kinds or kinds[kind]["ram_share"] is None: diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index 337b334..e126cf5 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/skills/kvb-org-queue/scripts/queue.py @@ -66,7 +66,9 @@ Config keys: budget_percent {ram, cpu}, required: the active and granted jobs' learned footprints (history.footprint, the whole budget for a kind not learned yet) may sum to at most these percentage points, a job alone always fits; slot_limits {kind: N}, optional, caps the learned slot limit of a kind from -above (history.slot_limits); queue_enforce (bool), optional, read by block +above (history.slot_limits); exclusive_kinds [kind] and light_kinds [kind], +optional: a job of an exclusive kind runs only beside jobs of light kinds, and +no job of another kind starts beside it; queue_enforce (bool), optional, read by block only. The watch.py keys disk, thresholds_gb (stop is the floor every grant leaves free), slot_need_gb (the disk seed per kind), load_limits (every grant also needs CPU busy below cpu_percent and RAM used plus the job's learned RAM @@ -349,7 +351,7 @@ def over_budget(used: dict[str, float], share: dict[str, float], budget: dict[st def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memory: str, silenced: bool, slot_limits: dict[str, int], footprints: dict[str, dict[str, float]], budget: dict[str, float], ranks: dict[str, int], agents: dict[str, int], loaded: dict[str, str], - now: float) -> dict[str, str | None]: + exclusive: set[str], light: set[str], now: float) -> dict[str, str | None]: """Decide every waiting entry in grant order, without disk or processes. Args: @@ -371,6 +373,8 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo all of them. loaded: {entry id: why live CPU or RAM holds it back (load_reason())} for the waiting entries the host load keeps out. + exclusive: kinds that run only beside light kinds (exclusive_kinds). + light: kinds allowed beside an exclusive one (light_kinds). now: epoch seconds; each whole hour waited moves an entry one rank up. Returns: @@ -379,7 +383,9 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo holds it, so a lower entry never takes a higher entry's slot; one blocked by disk or budget reserves its need against the entries after it. A hotfix entry that passes every check but the host load holds back - every entry after it, so lighter jobs cannot keep the load up. A waiting + every entry after it, so lighter jobs cannot keep the load up. An + exclusive entry that only running jobs of other kinds keep out holds + back every entry after it but light ones, so none overtakes it. A waiting entry whose holder has not polled for CLAIM_SECONDS reserves nothing and comes last. """ @@ -391,13 +397,20 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo busy = {e["workdir"] for e in active} verdicts: dict[str, str | None] = {} waiting = [e for e in entries if e["status"] == "waiting"] - held = None + held = drain = None for entry in ordered([e for e in waiting if not silent(e, now)], ranks, agents, now): if held is not None: verdicts[entry["id"]] = f"waits behind hotfix {held}, which the host load holds back" continue kind, need, workdir, share = entry["kind"], entry["need_gb"], entry["workdir"], footprints[entry["id"]] + if drain is not None and kind not in light: + verdicts[entry["id"]] = f"waits behind {drain}, which needs the host beside light kinds only" + continue reason = host_reason(stage, memory, silenced, need) or slot_reason(kind, workdir, slot_limits, counts, busy) + if reason is None: + reason = alone_reason(kind, counts, exclusive, light) + if reason is not None and kind in exclusive: + drain = entry["id"] if reason is None: reason, used, available = resources(need, share, used, available, floor, budget, taken) if reason is None and entry["hotfix"] and entry["id"] in loaded: @@ -464,6 +477,18 @@ def slot_reason(kind: str, workdir: str, slot_limits: dict[str, int], counts: Co return None +def alone_reason(kind: str, counts: Counter, exclusive: set[str], light: set[str]) -> str | None: + if kind in light: + return None + running = sorted(k for k, n in counts.items() if n and k not in light) + if kind in exclusive and running: + return f"{kind} runs only beside {', '.join(sorted(light)) or 'nothing'}, but {', '.join(running)} runs" + blocking = [k for k in running if k in exclusive] + if blocking: + return f"{', '.join(blocking)} runs only beside {', '.join(sorted(light)) or 'nothing'}" + return None + + def register_ranks() -> dict[str, int]: rasci = sibling("rasci") return {task: index for index, task in enumerate(rasci.ranked(rasci.merged(False)))} @@ -509,7 +534,8 @@ def decide(cfg: dict, entries: list[dict], free_gb: float, now: float, signature return evaluate( entries, free_gb, cfg["thresholds_gb"]["stop"], stage(cfg, free_gb, now), watch.memory_stage(watch.memory(), cfg), silenced(now), slots, footprints, budget, - register_ranks(), agent_ranks(), loaded, now, + register_ranks(), agent_ranks(), loaded, + set(cfg.get("exclusive_kinds", [])), set(cfg.get("light_kinds", [])), now, ) diff --git a/tests/test_queue.py b/tests/test_queue.py index 47f04e3..19701df 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -32,6 +32,8 @@ BASE = { "ranks": {"work/a": 0, "work/b": 1, "work/c": 2}, "agents": {}, "loaded": {}, + "exclusive": set(), + "light": set(), "now": NOW, } CHILD = ( @@ -201,6 +203,38 @@ class TestEvaluate(unittest.TestCase): self.assertEqual(list(self.decide(tied)), ["j5", "j6"]) self.assertEqual(list(self.decide(tied, agents={"Alpha": 0, "Beta": 1})), ["j6", "j5"]) + def test_an_exclusive_kind_runs_only_beside_light_kinds_in_both_directions(self): + gates = {"exclusive": {"quality-high", "test"}, "light": {"test-nostack"}} + slots = {"slot_limits": {"quality-high": 1, "iteration": 2, "test-nostack": 2}} + gate, deploy, light = job(1, kind="quality-high", status="active"), job(2, kind="iteration", workdir="/d"), job(3, kind="test-nostack", workdir="/l") + self.assertEqual(self.decide([gate, deploy, light], **slots), {"j2": None, "j3": None}) + self.assertEqual( + self.decide([gate, deploy, light], **slots, **gates), + {"j2": "quality-high runs only beside test-nostack", "j3": None}, + ) + running, waiting = job(2, kind="iteration", status="active", workdir="/d"), job(1, kind="quality-high") + self.assertEqual( + self.decide([running, waiting], **slots, **gates), + {"j1": "quality-high runs only beside test-nostack, but iteration runs"}, + ) + + def test_an_exclusive_entry_kept_out_by_a_running_job_holds_back_lower_non_light_entries(self): + gates = { + "exclusive": {"quality-high"}, "light": {"test-nostack"}, + "slot_limits": {"quality-high": 1, "iteration": 2, "test-nostack": 2}, + } + running = job(1, kind="iteration", status="active") + gate, deploy, light = job(2, kind="quality-high"), job(3, kind="iteration", task="work/b"), job(4, kind="test-nostack", task="work/c") + self.assertEqual(self.decide([running, gate, deploy, light], **gates), { + "j2": "quality-high runs only beside test-nostack, but iteration runs", + "j3": "waits behind j2, which needs the host beside light kinds only", + "j4": None, + }) + self.assertEqual( + self.decide([gate, deploy, light], **gates), + {"j2": None, "j3": "quality-high runs only beside test-nostack", "j4": None}, + ) + def test_the_slot_limit_counts_active_jobs_per_kind(self): verdicts = self.decide([job(1, kind="deploy", status="active"), job(2, kind="deploy"), job(3)]) self.assertEqual(verdicts, {"j2": "all 1 deploy slots taken", "j3": None}) @@ -283,6 +317,7 @@ class TestHistory(unittest.TestCase): ("python3", "-m", "unittest", "tests.test_queue"): "unittest tests.test_queue", ("python3", "-c", "print(1)"): "python3 -c", ("bash", "scripts/install.sh"): "install.sh", + ("python3", "/s/security.py", "--repo", "/w/.worktrees/fix/a", "--base", "HEAD"): "security.py", ("docker", "compose", "up", "-d"): "docker compose", ("/bin/true",): "true", } @@ -307,6 +342,12 @@ class TestHistory(unittest.TestCase): self.assertAlmostEqual(stats["disk_gb"], 7.6) self.assertEqual((stats["need_gb"], stats["minutes"]), (9, 7)) + def test_a_neighbours_cpu_spike_in_one_row_does_not_lift_the_learned_cpu_share(self): + rows = [*(history_row(n, cpu_before=10, cpu_max=12) for n in range(4)), history_row(4, cpu_before=10, cpu_max=90)] + signatures, kinds = self.learn(rows) + self.assertEqual((signatures["make test"]["cpu_share"], kinds["test"]["cpu_share"]), (2, 2)) + self.assertEqual(kinds["test"]["ram_share"], 10) + def test_learn_fits_the_slot_limit_between_the_median_load_and_the_load_limits(self): loads = ((10, 5), (20, 5), (20, 10), (30, 15), (40, 15)) rows = [history_row(n, mem_before=before, mem_max=before + share) for n, (before, share) in enumerate(loads)] @@ -682,7 +723,7 @@ class TestQueueCommands(unittest.TestCase): "| make compose-deploy | deploy | 5 | 12.0 | 4.6 | 10.0 | 6 |\n" "| true | test | 2 | - | - | - | - |\n" "\n" - "| Kind | Runs | p90 RAM share % | p90 CPU share % | Slot limit | Source |\n" + "| Kind | Runs | p90 RAM share % | Median CPU share % | Slot limit | Source |\n" "| --- | --- | --- | --- | --- | --- |\n" "| build | 0 | - | - | 1 | seed |\n" "| deploy | 5 | 10.0 | 20.0 | 5 | learned |\n"