From fb3c1ab158c8ffab258495138d1f1d21dc68b9a5 Mon Sep 17 00:00:00 2001 From: coolboy <131342541+wudixzy@users.noreply.github.com> Date: Fri, 2 Oct 2026 21:43:50 +0800 Subject: [PATCH] feat: pack recent submissions into reserved account slots first Preserve shared queue positions for old-model candidates with constrained admission. Retain headroom balancing for ties and weekly calibration pinning. Add offline mixed-age allocation regressions. --- README.md | 10 +++++++ modelhub_submmit_api/modelhub_client.py | 20 +++++++++++++ tests/test_concurrency.py | 37 +++++++++++++++++++++++++ 3 files changed, 67 insertions(+) diff --git a/README.md b/README.md index 258cea307a..2fcb461b12 100644 --- a/README.md +++ b/README.md @@ -311,6 +311,16 @@ tries the next account. It does not lower the persisted capacity using a cached active count: another worker may have filled the queue since that count was read. Downward capacity learning remains confined to weekly calibration. +Account allocation is age-aware: recent models first use accounts with only +reserved positions left, then the account with the fewest shared positions +remaining. Old models continue to use the account with the most eligible space. +Largest-headroom and rotating tie-breaks balance otherwise equivalent accounts; +weekly calibration remains pinned to its target account. For example, with +accounts at 94/100 and 95/100 and five recent-only positions each, assigning a +recent model to the second account preserves the first account's last old-model +position. A recent-then-old pair can both be accepted rather than stranding the +old model after the recent model takes its only eligible position. + Model verification results are reused for 15 minutes across poll cycles, and a locally failed model/GPU pair cools down for 24 hours instead of being excluded forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done` diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index a99a8fa967..05687d43ad 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -948,6 +948,26 @@ class ModelHubClientPool: usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0} if not usable: return None + if ( + reserve_capacity_slots is None + and weekly_target is None + and self.recent_model_reserve_slots > 0 + ): + # Recent models can use every position; old models cannot. + # Pack recent work into reserved positions first, then the + # account nearest its old-model threshold. This preserves + # shared positions for later old candidates, regardless of + # candidate arrival order. Keep largest-headroom/round-robin + # balancing among accounts with the same shared capacity. + shared_slots = { + index: max(0, remaining - self.recent_model_reserve_slots) + for index, remaining in usable.items() + } + least_shared = min(shared_slots.values()) + usable = { + index: remaining for index, remaining in usable.items() + if shared_slots[index] == least_shared + } best_remaining = max(usable.values()) tied = [index for index, remaining in usable.items() if remaining == best_remaining] selected_index = min( diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 98e3bc4653..e66e7ab5b5 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -217,6 +217,43 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): + def test_recent_work_preserves_shared_slots_for_later_old_models(self) -> None: + for initial_counts, ages, expected_counts in ( + ([94, 95], ["recent", "old"], [95, 96]), + ([93, 94], ["recent", "recent", "old", "old"], [95, 96]), + ): + with self.subTest(initial_counts=initial_counts): + clients = [FakeClient(active_count=count) for count in initial_counts] + pool = ModelHubClientPool( + clients, # type: ignore[arg-type] + active_task_cap=100, + recent_model_reserve_slots=5, + active_counts_ttl=60, + capacity_probe_interval_cycles=0, + ) + for index, age in enumerate(ages): + pool.add_task_for_model( + {"model": f"model-{index}"}, + model_last_modified=( + "2026-10-01T00:00:00Z" if age == "recent" + else "2025-01-01T00:00:00Z" + ), + submitted_at="2026-10-02T00:00:00Z", + ) + self.assertEqual(expected_counts, pool.active_task_counts()) + self.assertEqual(len(ages), sum(len(client.submitted) for client in clients)) + self.assertEqual(0, pool.old_model_submit_slots()) + + def test_recent_packing_keeps_largest_headroom_without_reserved_slots(self) -> None: + clients = [FakeClient(active_count=2), FakeClient(active_count=3)] + pool = ModelHubClientPool( + clients, # type: ignore[arg-type] + active_task_cap=5, + recent_model_reserve_slots=0, + ) + pool.add_task({"model": "recent"}) + self.assertEqual([1, 0], [len(client.submitted) for client in clients]) + def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: state_path = Path(temporary_dir) / "capacity.json"