diff --git a/src/svrf/daemon.py b/src/svrf/daemon.py index b2e96af..daa78f0 100644 --- a/src/svrf/daemon.py +++ b/src/svrf/daemon.py @@ -339,10 +339,6 @@ def _discover(self, requested, changed, state: dict, summary: dict, *, skip=froz self.stacked(row, heads, state, summary) else: summary["skipped"][str(n)] = decision - # A backlog (more candidates than admission slots) is read newest first, and a head whose read failed last - # tick after the rest, so old or unread heads never hold fresh admissions behind them; every candidate is - # still read this tick. Without a backlog every read starts at once and the order is kept. - bounded = bool(self.admission_round_reads) and len(candidates) > self.admission_round_reads if self.admission_round_reads: # every candidate waits from when it is first seen, and again from each read of it; a head that is no # longer a candidate (landed, closed, held) leaves the queue @@ -351,7 +347,17 @@ def _discover(self, requested, changed, state: dict, summary: dict, *, skip=froz for n in current: waiting.setdefault(n, self.clock()) state["waiting_since"] = waiting - if bounded or len(candidates) > max(1, self.admission_jobs): + return candidates + + def _order_candidates(self, candidates: list[dict], slots: int, state: dict, summary: dict) -> list[dict]: + """Order after observing capacity, before an unstarted head takes a free slot.""" + free = max(0, slots - len(self.inflight)) + if not free: + return candidates # no new read can start; the next discovery reads priority again + bounded = bool(self.admission_round_reads) and len(candidates) > self.admission_round_reads + # A backlog reads declared heads first, then the existing age or retry order. + # When every candidate fits, keep the list order without another command. + if bounded or len(candidates) > free: unread = {int(n) for n in (state.get("last_tick") or {}).get("retry_later") or [] if str(n).isdigit()} rank: dict[int, int] = {} if self.admission_priority_command: @@ -485,6 +491,7 @@ def _tick_body(self, requested: set[int] | None = None, changed: dict | None = N max_workers=self._pool_slots, thread_name_prefix="admission" ) slots = min(slots, self._pool_slots) + candidates = self._order_candidates(candidates, slots, state, summary) candidates = self._owned_candidates(candidates, slots, summary) if candidates: try: @@ -566,6 +573,7 @@ def _tick_body(self, requested: set[int] | None = None, changed: dict | None = N summary.pop("deferred", None) more = self._discover(requested, changed, state, summary, skip=set(self.inflight), seen=seen, refill=True) + more = self._order_candidates(more or [], slots, state, summary) # a declared head whose read failed this tick is not read again until the next tick lists: until # then no undeclared head is taken, so the declared lead the next tick instead of a whole backlog declared = (summary.get("admission_priority") or {}).get("read") or [] diff --git a/tests/test_admission_effective_priority.py b/tests/test_admission_effective_priority.py new file mode 100644 index 0000000..e1f2065 --- /dev/null +++ b/tests/test_admission_effective_priority.py @@ -0,0 +1,102 @@ +"""Declared work leads when current admission capacity leaves only one free slot.""" + +import tempfile +import threading +import unittest +from pathlib import Path + +from fakes import DaemonGitHub, DaemonRepo, FakeGate +from test_daemon import daemon + + +class EffectivePriority(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.release = threading.Event() + self.calls = [] + self.changed = threading.Condition() + self.owner = None + self.addCleanup(self.drain) + + def drain(self): + self.release.set() + if self.owner is not None: + while self.owner.inflight: + self.owner.tick(drain=True) + self.owner.close() + + def start(self, numbers, **options): + self.repo = DaemonRepo(numbers) + self.gh = DaemonGitHub(self.repo) + + def admission(head, base): + with self.changed: + self.calls.append(head) + self.changed.notify_all() + if not self.release.wait(10): + raise RuntimeError('test admission was not released') + return {'verdict': 'MERGEABLE', 'residuals': []} + + self.owner = daemon( + self.repo, self.gh, FakeGate(self.repo), admission, Path(self.temp.name), + jobs=2, admission_jobs=4, admission_round_reads=8, tick_seconds=0.05, + admission_priority_command='echo 42', **options, + ) + + def await_calls(self, count): + with self.changed: + self.assertTrue(self.changed.wait_for(lambda: len(self.calls) >= count, 2)) + + def open_head(self, number): + sha = f'h{number}' + self.repo.commits[sha] = {'lanes': frozenset({0, number}), 'parents': ('c0',)} + self.repo.heads[number] = sha + self.repo.branches[f'pr-{number}'] = sha + + def test_declared_head_gets_the_slot_left_by_running_admissions(self): + self.start([1, 2, 3]) + first = self.owner.tick() + self.await_calls(3) + self.assertEqual(first['inflight'], [1, 2, 3]) + original = {number: item['future'] for number, item in self.owner.inflight.items()} + self.open_head(41) + self.open_head(42) + + second = self.owner.tick() + self.await_calls(4) + + self.assertEqual(self.calls[3], 'h42') + self.assertEqual(second['inflight'], [1, 2, 3, 42]) + self.assertEqual(second['deferred'], [41]) + self.assertEqual(second['admission_priority']['read'], [42]) + for number, future in original.items(): + self.assertIs(self.owner.inflight[number]['future'], future) + self.assertEqual(len(self.calls), 4) + + def test_declared_head_leads_at_a_lower_observed_capacity(self): + self.start([41, 42], admission_capacity_command='echo 1') + + out = self.owner.tick() + self.await_calls(1) + + self.assertEqual(self.calls, ['h42']) + self.assertEqual(out['inflight'], [42]) + self.assertEqual(out['deferred'], [41]) + self.assertEqual(out['admission_capacity']['slots'], 1) + self.assertEqual(out['admission_priority']['read'], [42]) + + def test_a_queue_that_fits_starts_without_a_priority_read(self): + self.start([41, 42]) + + out = self.owner.tick() + self.await_calls(2) + + self.assertCountEqual(self.calls, ['h41', 'h42']) + self.assertEqual(out['inflight'], [41, 42]) + self.assertNotIn('admission_priority', out) + self.assertNotIn('deferred', out) + + +if __name__ == '__main__': + unittest.main()