Merge fix/kvb-org-queue-footprint-noise into master

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
+4. An unblocked job counts as granted for the jobs behind it.
-   slot.
 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
+- per kind, the slot limit is how many p90 RAM and median CPU shares fit
-  median load at the grant and `load_limits`, at least 1;
+  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 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

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:
       <first operand>` for an interpreter running a script (python, bash, ...)
       and `<interpreter> -c` for inline code; else `<argv[0] basename> <first
       operand>`. 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
+      and runs is None. minutes is the median, disk_gb and ram_share the p90
-      cpu_share the p90 of the disk taken and of max minus before in
+      of the disk taken and of max minus before in percentage points,
-      percentage points; need_gb is ceil(disk_gb + DISK_MARGIN_GB); limit is
+      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:

skills/kvb-org-queue/scripts/queue.py

@@ -70,7 +70,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
@@ -372,7 +374,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:
@@ -394,6 +396,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:
@@ -402,7 +406,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.
     """
@@ -414,13 +420,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:
@@ -487,6 +500,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)))}
@@ -532,7 +557,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,
     )
 
 

tests/test_queue.py

@@ -33,6 +33,8 @@ BASE = {
     "ranks": {"work/a": 0, "work/b": 1, "work/c": 2},
     "agents": {},
     "loaded": {},
+    "exclusive": set(),
+    "light": set(),
     "now": NOW,
 }
 CHILD = (
@@ -202,6 +204,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})
@@ -284,6 +318,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",
         }
@@ -308,6 +343,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)]
@@ -718,7 +759,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"