Merge fix/kvb-org-queue/unclaimed-grant into master

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)

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)