unsloth/studio/backend/tests/test_gpu_arbiter.py
Daniel Han 899465ed80 Studio: close arbiter load-registration race and surface native progress + local pipeline folders
Publish native sd.cpp generate progress (_gen) before LoRA resolution so a reload probe reads active during setup, matching the diffusers path.

Register the diffusion/video GPU load under the arbiter lock (acquire_for now takes a register callback) so a competing acquire cannot evict an owner before its load is marked in-flight and let two loaders allocate VRAM at once.

Admit local diffusers pipeline folders (root model_index.json, weights in component subdirs) in the local model scan so they reach task tagging and the On Device picker.
2026-07-13 06:31:00 +00:00

178 lines
5.8 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Unit tests for the single-GPU arbiter.
The real evictors (which tear down live backends) are replaced with recorders, so
these verify only the ownership/eviction sequencing — no torch, GPU, or subprocess.
"""
from __future__ import annotations
import pytest
import core.inference.gpu_arbiter as arb
@pytest.fixture
def calls(monkeypatch):
recorded: list[str] = []
monkeypatch.setattr(arb, "_owner", None)
monkeypatch.setitem(arb._EVICTORS, arb.CHAT, lambda: recorded.append("evict-chat"))
monkeypatch.setitem(arb._EVICTORS, arb.DIFFUSION, lambda: recorded.append("evict-diffusion"))
return recorded
def test_first_acquire_evicts_nothing(calls):
arb.acquire_for(arb.CHAT)
assert calls == []
assert arb.current_owner() == arb.CHAT
def test_diffusion_load_evicts_chat(calls):
arb.acquire_for(arb.CHAT)
arb.acquire_for(arb.DIFFUSION)
assert calls == ["evict-chat"]
assert arb.current_owner() == arb.DIFFUSION
def test_chat_load_evicts_diffusion(calls):
arb.acquire_for(arb.DIFFUSION)
arb.acquire_for(arb.CHAT)
assert calls == ["evict-diffusion"]
assert arb.current_owner() == arb.CHAT
def test_reacquiring_same_owner_does_not_evict(calls):
arb.acquire_for(arb.CHAT)
arb.acquire_for(arb.CHAT)
assert calls == []
assert arb.current_owner() == arb.CHAT
def test_release_clears_owner(calls):
arb.acquire_for(arb.DIFFUSION)
arb.release(arb.DIFFUSION)
assert arb.current_owner() is None
# A subsequent chat load then has nothing to evict.
arb.acquire_for(arb.CHAT)
assert calls == []
def test_release_by_non_owner_is_noop(calls):
arb.acquire_for(arb.CHAT)
arb.release(arb.DIFFUSION)
assert arb.current_owner() == arb.CHAT
def test_unknown_owner_raises(calls):
with pytest.raises(ValueError):
arb.acquire_for("gpu")
def test_evict_chat_unloads_a_still_loading_chat_backend(monkeypatch):
# A chat model still starting up is is_active (process exists) but not yet
# is_loaded (healthy). Eviction must still unload it, or the load would keep
# allocating VRAM after the GPU was handed to diffusion.
import core.inference as core_inference
import routes.inference as routes_inference
unloaded: list[bool] = []
class _FakeLlama:
is_active = True
is_loaded = False # still loading: skipped if eviction gates on is_loaded
def unload_model(self):
unloaded.append(True)
def _wait_for_vram_settle(self, *, since_kill):
pass
class _FakeOrchestrator:
active_model_name = None
def unload_model(self, name):
pass
def _shutdown_subprocess(self, timeout = 5.0):
pass
monkeypatch.setattr(routes_inference, "get_llama_cpp_backend", lambda: _FakeLlama())
monkeypatch.setattr(core_inference, "get_inference_backend", lambda: _FakeOrchestrator())
arb._evict_chat()
assert unloaded == [True] # still-loading chat backend was unloaded, not skipped
def test_register_runs_under_ownership_and_returns_result(calls):
# A register callback runs after ownership transfers (owner already set) and its
# return value is forwarded -- the route uses this to register the in-flight load.
seen_owner: list = []
def register():
seen_owner.append(arb.current_owner())
return "status-dict"
result = arb.acquire_for(arb.DIFFUSION, register)
assert result == "status-dict"
assert seen_owner == [arb.DIFFUSION]
assert arb.current_owner() == arb.DIFFUSION
def test_register_failure_leaves_ownership_in_place(calls):
# A failing register (e.g. begin_load reporting a load already in progress) propagates
# but must not drop ownership -- the prior handoff (chat already evicted) stands.
arb.acquire_for(arb.CHAT)
def register():
raise RuntimeError("A diffusion load is already in progress.")
with pytest.raises(RuntimeError):
arb.acquire_for(arb.DIFFUSION, register)
assert calls == ["evict-chat"]
assert arb.current_owner() == arb.DIFFUSION
def test_competing_acquire_blocks_until_register_completes(monkeypatch):
# The window this closes: while DIFFUSION registers its load, a competing VIDEO acquire
# must not evict DIFFUSION until the load is marked in-flight. Holding the lock across
# register makes the competitor wait, so eviction never races an unregistered load.
import threading
import time
monkeypatch.setattr(arb, "_owner", None)
evicted: list = []
monkeypatch.setitem(arb._EVICTORS, arb.DIFFUSION, lambda: evicted.append("evict-diffusion"))
monkeypatch.setitem(arb._EVICTORS, arb.VIDEO, lambda: evicted.append("evict-video"))
in_register = threading.Event()
release_register = threading.Event()
def register():
in_register.set()
# Hold the arbiter lock here; a competing acquire_for(VIDEO) must block until we return.
assert release_register.wait(2.0)
return "loading"
loader = threading.Thread(target = lambda: arb.acquire_for(arb.DIFFUSION, register))
loader.start()
assert in_register.wait(2.0)
competitor_done = threading.Event()
threading.Thread(
target = lambda: (arb.acquire_for(arb.VIDEO), competitor_done.set()),
).start()
# The competitor cannot evict DIFFUSION while register still holds the lock.
time.sleep(0.1)
assert evicted == []
assert not competitor_done.is_set()
# Let register finish; ownership is now safely registered, so the competitor proceeds.
release_register.set()
loader.join(2.0)
assert competitor_done.wait(2.0)
assert evicted == ["evict-diffusion"]
assert arb.current_owner() == arb.VIDEO