diff --git a/scripts/ci/late_placement.py b/scripts/ci/late_placement.py index eba926cd7816..c4aba87b4563 100644 --- a/scripts/ci/late_placement.py +++ b/scripts/ci/late_placement.py @@ -31,7 +31,7 @@ Blacksmith (about 350 s against 240 to 400 s), so waiting for a mini buys nothing. When the picker owned some of this run's gui-token jobs and the gui runners idle now cannot take them all, this counts the gui-label jobs already -queued (gui_backlog(): the jobs of in-flight CI runs, newest runs first, +queued (gui_backlog(): the jobs of in-flight CI runs, oldest runs first, stopping once the answer cannot change). The run keeps on the gui label only the jobs that start within GUI_QUEUE_ROUNDS gui job lengths (the idle runners plus that many rounds of the online ones, less the backlog), and gives the @@ -44,6 +44,7 @@ """ from __future__ import annotations +import concurrent.futures import datetime as dt import importlib.util import json @@ -71,6 +72,7 @@ def _picker(): # jobs queue only once its admission finished (p50 about 9 minutes), so a run younger than BACKLOG_MIN_AGE has none, # and one older than the window has finished its shards. BACKLOG_LOOKUPS = 30 +BACKLOG_READERS = 8 BACKLOG_WINDOW_MINUTES = 120 BACKLOG_MIN_AGE_MINUTES = 4 @@ -108,7 +110,8 @@ def place(jobs: Sequence[str], *, owned_jobs: str, idle: int, root: str, gui: bo def gui_backlog(github: Any, label: str, *, exclude_run_id: int | None, enough: int, now: dt.datetime) -> int: - """Jobs queued on `label` in the CI runs still in flight, stopping at `enough` (one request per run). + """Jobs queued on `label` in the CI runs still in flight (one request per run), read BACKLOG_READERS runs + at a time and stopping after the batch that reaches `enough`, so the count may pass it. GitHub lists a run as `queued` while any of its jobs is, even with others running, so both `queued` and `in_progress` runs are read. Oldest first, @@ -123,13 +126,19 @@ def gui_backlog(github: Any, label: str, *, exclude_run_id: int | None, enough: created = pool.parse_time(str(run.get("created_at") or "")) if run.get("id") != exclude_run_id and created is not None and created <= newest: runs[run.get("id")] = run - queued = 0 - for run in sorted(runs.values(), key=lambda run: str(run.get("created_at")))[:BACKLOG_LOOKUPS]: - if queued >= enough: - break + def queued_in(run: Mapping[str, Any]) -> int: jobs = github.get(f"/actions/runs/{run['id']}/jobs?filter=latest&per_page={pool.PAGE_SIZE}").get("jobs") or [] - queued += sum(1 for job in jobs if isinstance(job, Mapping) and job.get("status") == "queued" - and label in (job.get("labels") or [])) + return sum(1 for job in jobs if isinstance(job, Mapping) and job.get("status") == "queued" + and label in (job.get("labels") or [])) + + # Read BACKLOG_READERS runs at a time: one by one, 30 job lists under load outran the step's minute. + ordered = sorted(runs.values(), key=lambda run: str(run.get("created_at")))[:BACKLOG_LOOKUPS] + queued = 0 + with concurrent.futures.ThreadPoolExecutor(max_workers=BACKLOG_READERS) as readers: + for start in range(0, len(ordered), BACKLOG_READERS): + if queued >= enough: + break + queued += sum(readers.map(queued_in, ordered[start:start + BACKLOG_READERS])) return queued diff --git a/tests/test_ci_late_placement.py b/tests/test_ci_late_placement.py index fd2da95e0033..97fded7a5c9b 100644 --- a/tests/test_ci_late_placement.py +++ b/tests/test_ci_late_placement.py @@ -230,11 +230,55 @@ def get(self, path): return {"jobs": [{"status": "queued", "labels": [GUI]}, {"status": "queued", "labels": [ROOT_STD]}, {"status": "in_progress", "labels": [GUI]}, {"status": "queued", "labels": [GUI]}]} api = API() - self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=7, enough=5, now=now), 6) - # Oldest first; run 4 is too young to have gui jobs and 7 is this run. - self.assertEqual(api.jobs_read, [8, 6, 5]) + # One batch of readers takes every eligible run, so the count may pass `enough` (decide() only + # compares against it). Run 4 is too young to have gui jobs and 7 is this run. + self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=7, enough=5, now=now), 8) + self.assertEqual(sorted(api.jobs_read), [5, 6, 8, 9]) self.assertEqual(api.statuses, [("ci.yml", "queued"), ("ci.yml", "in_progress")]) + def test_backlog_reads_runs_concurrently_oldest_batch_first_and_stops_when_enough(self): + import datetime as dt + import threading + now = dt.datetime(2026, 9, 28, 1, 0, tzinfo=dt.timezone.utc) + runs = [{"id": i, "created_at": (now - dt.timedelta(minutes=100 - i)).strftime("%Y-%m-%dT%H:%M:%SZ")} + for i in range(1, 21)] + # Every read of a batch must be in flight at once, or the barrier breaks and the read raises. + together = threading.Barrier(late.BACKLOG_READERS, timeout=30) + + class API: + def __init__(self): + self.jobs_read, self.lock = [], threading.Lock() + + def runs_since(self, workflow, since, **filters): + return runs if filters["status"] == "in_progress" else [] + + def get(self, path): + together.wait() + with self.lock: + self.jobs_read.append(int(path.split("/")[3])) + return {"jobs": [{"status": "queued", "labels": [GUI]}]} + api = API() + self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=None, enough=late.BACKLOG_READERS, now=now), + late.BACKLOG_READERS) + # One batch: the oldest BACKLOG_READERS runs, and nothing after `enough`. + self.assertEqual(sorted(api.jobs_read), list(range(1, late.BACKLOG_READERS + 1))) + + def test_a_failed_backlog_read_raises_so_nothing_moves(self): + import datetime as dt + now = dt.datetime(2026, 9, 28, 1, 0, tzinfo=dt.timezone.utc) + + class API: + def runs_since(self, workflow, since, **filters): + return [{"id": i, "created_at": (now - dt.timedelta(minutes=30 + i)).strftime("%Y-%m-%dT%H:%M:%SZ")} + for i in range(1, 4)] if filters["status"] == "in_progress" else [] + + def get(self, path): + if path.split("/")[3] == "2": + raise RuntimeError("HTTP 502") + return {"jobs": []} + with self.assertRaises(RuntimeError): + late.gui_backlog(API(), GUI, exclude_run_id=None, enough=5, now=now) + class Output(unittest.TestCase): def test_main_writes_an_empty_object_without_a_token(self):