diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index dc25240..89bffde 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/skills/kvb-org-queue/scripts/queue.py @@ -120,6 +120,7 @@ SIBLINGS = { "nofollow": SKILLS / "kvb-org-organize" / "scripts" / "nofollow.py", } POLL_SECONDS = 20 +CLAIM_SECONDS = 3 * POLL_SECONDS TICK_SECONDS = 0.2 HOUR = 3600 CANCELLED = 3 @@ -373,6 +374,8 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo An unblocked entry counts as granted for the entries after it, whoever 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 waiting entry whose holder has not polled for CLAIM_SECONDS reserves + nothing and comes last. """ active = [e for e in entries if e["status"] == "active"] available = free_gb - sum(e["need_gb"] for e in active) @@ -381,7 +384,8 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo counts = Counter(e["kind"] for e in active) busy = {e["workdir"] for e in active} verdicts: dict[str, str | None] = {} - for entry in ordered([e for e in entries if e["status"] == "waiting"], ranks, agents, now): + waiting = [e for e in entries if e["status"] == "waiting"] + for entry in ordered([e for e in waiting if not silent(e, now)], ranks, agents, now): kind, need, workdir, share = entry["kind"], entry["need_gb"], entry["workdir"], footprints[entry["id"]] reason = host_reason(stage, memory, silenced, need) or slot_reason(kind, workdir, slot_limits, counts, busy) if reason is None: @@ -391,9 +395,23 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo counts[kind] += 1 busy.add(workdir) verdicts[entry["id"]] = reason + for entry in waiting: + if silent(entry, now): + verdicts[entry["id"]] = f"no poll for {now - entry['polled']:.0f} s; reserves nothing until it polls again" return verdicts +def silent(entry: dict, now: float) -> bool: + return entry["status"] == "waiting" and now - entry.get("polled", now) > CLAIM_SECONDS + + +def rejoin(data: dict, entry: dict, now: float) -> None: + if silent(entry, now): + data["seq"] += 1 + entry.update(seq=data["seq"], enqueued=now) + entry["polled"] = now + + def plus(used: dict[str, float], share: dict[str, float]) -> dict[str, float]: return {resource: used[resource] + share[resource] for resource in used} @@ -572,6 +590,7 @@ def wait(cfg: dict, entry_id: str, caught: list[int]) -> tuple[int | None, int]: if not mine: print(f"queue: {entry_id} was cancelled", file=sys.stderr) return CANCELLED, 0 + rejoin(data, mine[0], now) verdicts = decide(cfg, data["entries"], free, now, signatures, kinds) pending = pending_ram(data["entries"], entry_id, now, signatures, kinds) reason = verdicts[entry_id] or load_reason(cfg["load_limits"], cpu, mem, pending) diff --git a/tests/test_queue.py b/tests/test_queue.py index f047fba..f969485 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -208,6 +208,22 @@ class TestEvaluate(unittest.TestCase): verdicts = self.decide([job(1, task="work/b", kind="deploy"), job(2, kind="deploy")]) self.assertEqual(verdicts, {"j2": None, "j1": "all 1 deploy slots taken"}) + def test_a_waiter_that_stopped_polling_reserves_nothing_and_the_next_one_is_granted(self): + behind = job(3, task="work/b", kind="deploy", polled=NOW) + verdicts = self.decide([job(2, kind="deploy", polled=NOW - 61), behind]) + self.assertEqual(verdicts, {"j3": None, "j2": "no poll for 61 s; reserves nothing until it polls again"}) + for polling in (job(2, kind="deploy", polled=NOW - 59), job(2, kind="deploy")): + self.assertEqual(self.decide([polling, behind]), {"j2": None, "j3": "all 1 deploy slots taken"}) + + def test_a_silent_waiter_rejoins_at_the_back_once_it_polls_again(self): + data = {"seq": 7, "entries": []} + silent = job(2, polled=NOW - 61, enqueued=NOW - 2 * HOUR) + self.queue.rejoin(data, silent, NOW) + self.assertEqual((silent["seq"], silent["enqueued"], silent["polled"], data["seq"]), (8, NOW, NOW, 8)) + polling = job(3, polled=NOW - 30, enqueued=NOW - HOUR) + self.queue.rejoin(data, polling, NOW) + self.assertEqual((polling["seq"], polling["enqueued"], polling["polled"], data["seq"]), (3, NOW - HOUR, NOW, 8)) + def test_disk_reservation_keeps_smaller_jobs_from_starving_a_big_one(self): big, small = job(1, need_gb=50), job(2, task="work/b", need_gb=5) verdicts = self.decide([big, small], free_gb=40.0)