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.
This commit is contained in:
10
README.md
10
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
|
active count: another worker may have filled the queue since that count was
|
||||||
read. Downward capacity learning remains confined to weekly calibration.
|
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
|
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
|
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`
|
forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done`
|
||||||
|
|||||||
@@ -948,6 +948,26 @@ class ModelHubClientPool:
|
|||||||
usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0}
|
usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0}
|
||||||
if not usable:
|
if not usable:
|
||||||
return None
|
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())
|
best_remaining = max(usable.values())
|
||||||
tied = [index for index, remaining in usable.items() if remaining == best_remaining]
|
tied = [index for index, remaining in usable.items() if remaining == best_remaining]
|
||||||
selected_index = min(
|
selected_index = min(
|
||||||
|
|||||||
@@ -217,6 +217,43 @@ def make_candidate(index: int) -> dict:
|
|||||||
|
|
||||||
|
|
||||||
class ClientPoolConcurrencyTests(unittest.TestCase):
|
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:
|
def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None:
|
||||||
with tempfile.TemporaryDirectory() as temporary_dir:
|
with tempfile.TemporaryDirectory() as temporary_dir:
|
||||||
state_path = Path(temporary_dir) / "capacity.json"
|
state_path = Path(temporary_dir) / "capacity.json"
|
||||||
|
|||||||
Reference in New Issue
Block a user