unsloth/studio/backend/tests/test_tensor_parallel.py
Daniel Han 9c2eacc35e
Studio: reserve CUDA context and mmproj/MTP soft overhead in the GGUF fit budget (#6718)
---------

Co-authored-by: Lee Jackson <130007945+Imagineer99@users.noreply.github.com>
Co-authored-by: oobabooga <112222186+oobabooga@users.noreply.github.com>
2026-07-03 13:07:30 -03:00

1234 lines
48 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
"""Backend contract for the Tensor Parallelism toggle.
The toggle threads a single ``tensor_parallel`` bool from the chat UI
through the load request to a ``--split-mode tensor`` llama-server flag,
and round-trips it back via the load/status responses so the switch
reflects what is actually running. These tests pin:
* the pydantic request/response/status contract (snake_case key,
default False),
* the backend ``tensor_parallel`` property and its reset on unload,
* the ``_already_in_target_state`` reload-detection branch, and
* that ``--split-mode tensor`` is emitted only behind the toggle.
"""
from __future__ import annotations
import asyncio
import inspect
import sys
import threading
import time
import types as _types
from pathlib import Path
import pytest
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
if _BACKEND_DIR not in sys.path:
sys.path.insert(0, _BACKEND_DIR)
# Same external-dep stubs as the other llama_cpp unit tests so importing
# the backend doesn't drag in structlog / httpx / loggers.
_loggers_stub = _types.ModuleType("loggers")
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
sys.modules.setdefault("loggers", _loggers_stub)
_structlog_stub = _types.ModuleType("structlog")
_structlog_stub.get_logger = lambda *a, **k: __import__("logging").getLogger("stub")
sys.modules.setdefault("structlog", _structlog_stub)
_httpx_stub = _types.ModuleType("httpx")
for _exc in (
"ConnectError",
"TimeoutException",
"ReadTimeout",
"ReadError",
"RemoteProtocolError",
"CloseError",
):
setattr(_httpx_stub, _exc, type(_exc, (Exception,), {}))
_httpx_stub.Timeout = type("T", (), {"__init__": lambda s, *a, **k: None})
_httpx_stub.Client = type(
"C",
(),
{
"__init__": lambda s, **kw: None,
"__enter__": lambda s: s,
"__exit__": lambda s, *a: None,
},
)
sys.modules.setdefault("httpx", _httpx_stub)
from core.inference import llama_cpp as llama_cpp_module
from core.inference.llama_cpp import _CTX_FIT_VRAM_FRACTION, LlamaCppBackend
from core.inference.llama_server_args import (
_effective_tensor_parallel,
resolve_tensor_parallel,
)
from core.inference.tensor_fallback import load_with_tensor_fallback
from models.inference import (
InferenceStatusResponse,
LoadRequest,
LoadResponse,
)
# ── Pydantic contract (snake_case key, default False) ────────────────
def test_load_request_defaults_tensor_parallel_false():
req = LoadRequest(model_path = "owner/repo")
assert req.tensor_parallel is False
def test_load_request_accepts_tensor_parallel():
req = LoadRequest(model_path = "owner/repo", tensor_parallel = True)
assert req.tensor_parallel is True
def test_load_request_round_trips_json_key():
# The frontend sends the snake_case key verbatim.
req = LoadRequest.model_validate({"model_path": "owner/repo", "tensor_parallel": True})
assert req.tensor_parallel is True
assert req.model_dump()["tensor_parallel"] is True
@pytest.mark.parametrize("model_cls", [LoadResponse, InferenceStatusResponse])
def test_response_models_emit_tensor_parallel(model_cls):
# Default False, and the key is always present in the JSON body.
if model_cls is LoadResponse:
default = model_cls(
status = "loaded",
model = "owner/repo",
display_name = "repo",
inference = {},
)
on = model_cls(
status = "loaded",
model = "owner/repo",
display_name = "repo",
inference = {},
tensor_parallel = True,
)
else:
default = model_cls()
on = model_cls(tensor_parallel = True)
assert default.model_dump()["tensor_parallel"] is False
assert on.model_dump()["tensor_parallel"] is True
# ── Backend property + reset ─────────────────────────────────────────
class _FakeProcess:
"""Stand-in for subprocess.Popen so _kill_process is a no-op."""
def terminate(self):
pass
def wait(self, timeout = None):
return 0
def kill(self):
pass
def poll(self):
return 0
def test_tensor_parallel_property_defaults_false():
assert LlamaCppBackend().tensor_parallel is False
def test_tensor_parallel_property_reflects_field():
backend = LlamaCppBackend()
backend._tensor_parallel = True
assert backend.tensor_parallel is True
def test_unload_resets_tensor_parallel():
backend = LlamaCppBackend()
backend._process = _FakeProcess()
backend._tensor_parallel = True
backend.unload_model()
assert backend.tensor_parallel is False
# ── _already_in_target_state reload-detection branch ─────────────────
def _loaded_backend(tensor_parallel: bool) -> LlamaCppBackend:
backend = LlamaCppBackend()
backend._process = _FakeProcess() # is_loaded only checks "is not None"
backend._healthy = True
backend._model_identifier = "owner/repo"
backend._hf_variant = "Q4_K_M"
backend._requested_n_ctx = 8192
backend._cache_type_kv = None
backend._requested_spec_mode = "auto"
backend._chat_template_override = None
backend._is_vision = False
backend._extra_args = None
backend._gguf_path = None
backend._tensor_parallel = tensor_parallel
return backend
def _target_state(backend: LlamaCppBackend, tensor_parallel: bool) -> bool:
return backend._already_in_target_state(
gguf_path = None,
model_identifier = "owner/repo",
hf_variant = "Q4_K_M",
n_ctx = 8192,
cache_type_kv = None,
speculative_type = "auto",
chat_template_override = None,
extra_args = None,
is_vision = False,
tensor_parallel = tensor_parallel,
)
@pytest.mark.parametrize("flag", [True, False])
def test_already_in_target_state_matches_same_tensor_parallel(flag):
assert _target_state(_loaded_backend(flag), flag) is True
@pytest.mark.parametrize(
"loaded,requested",
[(False, True), (True, False)],
)
def test_already_in_target_state_reloads_on_tensor_parallel_change(loaded, requested):
# Flipping the toggle either direction must force a reload so the
# command is rebuilt with/without --split-mode tensor.
assert _target_state(_loaded_backend(loaded), requested) is False
def test_already_in_target_state_reconciles_split_mode_extras():
# Tensor engaged via --split-mode in extras (boolean omitted/default False)
# must match a server already running tensor mode -- no spurious reload.
backend = _loaded_backend(tensor_parallel = True)
backend._extra_args = ["--split-mode", "tensor"]
assert (
backend._already_in_target_state(
gguf_path = None,
model_identifier = "owner/repo",
hf_variant = "Q4_K_M",
n_ctx = 8192,
cache_type_kv = None,
speculative_type = "auto",
chat_template_override = None,
extra_args = ["--split-mode", "tensor"],
is_vision = False,
tensor_parallel = False,
)
is True
)
# ── --split-mode tensor is emitted only behind the toggle ────────────
def _load_model_source() -> str:
return inspect.getsource(llama_cpp_module.LlamaCppBackend.load_model)
def test_split_mode_tensor_is_gated_on_the_toggle():
src = _load_model_source()
assert (
'cmd.extend(["--split-mode", "tensor"])' in src
), "the tensor-parallel flag emission must be present in load_model"
# The emission lives behind `if tensor_parallel:` -- it must never be
# part of the unconditional base cmd list.
base_start = src.find("cmd = [")
base_end = src.find("\n ]", base_start)
base_block = src[base_start:base_end] if base_end > base_start else ""
assert (
"--split-mode" not in base_block
), "--split-mode must be conditional, not in the base cmd list"
gate = src.find("if tensor_parallel:")
emit = src.find('cmd.extend(["--split-mode", "tensor"])')
assert 0 <= gate < emit, "emission must sit under `if tensor_parallel:`"
def test_proportional_tensor_split_is_emitted_in_tensor_mode():
# Asymmetric GPUs (e.g. 48 GB + 24 GB) OOM the smaller card under the
# even default; the allocator weights --tensor-split by free VRAM. Pin
# that the flag is emitted from inside the tensor-parallel block.
src = _load_model_source()
assert '"--tensor-split"' in src
gate = src.find("if tensor_parallel:")
ts = src.find('"--tensor-split"')
nxt_else = src.find("self._tensor_parallel = False")
assert 0 <= gate < ts < nxt_else, "--tensor-split must be emitted under `if tensor_parallel:`"
def test_mtp_decode_probe_wired_under_tensor_parallel():
# MTP-draft can pass /health and crash the CUDA FA kernel only on the first
# decode under --split-mode tensor. Rather than statically banning MTP+TP
# (which a future llama.cpp may support), load_model probes a decode; a hard
# fault retries --flash-attn off first, else routes into the MTP-drop fallback.
src = _load_model_source()
probe = src.find("_probe_mtp_decode()")
assert probe != -1, "load_model must decode-probe MTP under tensor parallelism"
# Gated on tensor mode AND an MTP request (ordinary MTP loads stay unprobed).
guard = src[max(0, probe - 400) : probe]
assert "self._tensor_parallel" in guard and "_spec_requested_mtp" in guard
# A hard fault retries FA-off (keeps MTP) before flipping healthy so the
# shared MTP-drop fallback fires.
after = src[probe : probe + 900]
assert "_with_flash_attn_off" in after and "healthy = False" in after
fallback = src.find("if not healthy and _spec_requested_mtp")
assert 0 <= probe < fallback, "the probe must precede the MTP-drop fallback"
def test_probe_mtp_decode_returns_false_on_crash(monkeypatch):
# The probe is the decode-time health gate: True only on a clean 200 from a
# live server; any error (dropped connection, non-200, dead process) is a
# failed probe so the caller drops MTP and retries.
backend = LlamaCppBackend()
backend._port = 0
class _Resp:
def __init__(self, code):
self.status_code = code
backend._process = None # liveness check skipped; exercise the HTTP result
monkeypatch.setattr(llama_cpp_module.httpx, "post", lambda *a, **k: _Resp(200), raising = False)
assert backend._probe_mtp_decode(timeout = 1.0) is True
def _drop(*a, **k):
raise llama_cpp_module.httpx.RemoteProtocolError("peer closed connection")
monkeypatch.setattr(llama_cpp_module.httpx, "post", _drop, raising = False)
assert backend._probe_mtp_decode(timeout = 1.0) is False
monkeypatch.setattr(llama_cpp_module.httpx, "post", lambda *a, **k: _Resp(500), raising = False)
assert backend._probe_mtp_decode(timeout = 1.0) is False
# 200 but the server aborted right after (poll() returns an exit code).
backend._process = _FakeProcess()
monkeypatch.setattr(llama_cpp_module.httpx, "post", lambda *a, **k: _Resp(200), raising = False)
assert backend._probe_mtp_decode(timeout = 1.0) is False
# ── generation-time MTP recovery (mid-stream crash) ──────────────────
def _recovery_backend() -> LlamaCppBackend:
# A backend that loaded MTP under tensor parallelism and whose server has
# since exited (the _FakeProcess poll() returns 0 -> a dead subprocess).
b = LlamaCppBackend()
b._tensor_parallel = True
b._speculative_type = "draft-mtp"
b._mtp_runtime_fallback_active = True
b._process = _FakeProcess()
b._last_load_kwargs = {
"model_identifier": "owner/repo",
"tensor_parallel": True,
"speculative_type": "auto",
"n_parallel": 4,
}
return b
def test_generate_chat_completion_wires_runtime_recovery():
# The non-tool generation path must route a mid-stream server death into the
# recovery helper (the tool + passthrough paths do so from the routes).
src = inspect.getsource(LlamaCppBackend.generate_chat_completion)
assert "_maybe_recover_from_mtp_crash" in src
def test_runtime_recovery_reloads_without_mtp(monkeypatch):
# One background reload with speculative_type="off" (rest of snapshot kept),
# then spec_fallback_reason="runtime_error" and single-flight released.
b = _recovery_backend()
done = threading.Event()
captured = {}
def _fake_load_model(**kwargs):
captured.update(kwargs)
done.set()
return True
monkeypatch.setattr(b, "load_model", _fake_load_model)
assert b._maybe_recover_from_mtp_crash(RuntimeError("peer closed")) is True
assert done.wait(timeout = 5)
assert captured["speculative_type"] == "off"
assert captured["model_identifier"] == "owner/repo"
assert captured["n_parallel"] == 4 # snapshot replayed faithfully
deadline = time.monotonic() + 2
while b._spec_fallback_reason != "runtime_error" and time.monotonic() < deadline:
time.sleep(0.02)
assert b._spec_fallback_reason == "runtime_error"
# The reload thread clears the single-flight flag in its finally, a beat after
# it sets the fallback reason -- wait for that instead of racing the thread.
deadline = time.monotonic() + 2
while b._mtp_runtime_fallback_in_progress and time.monotonic() < deadline:
time.sleep(0.02)
assert b._mtp_runtime_fallback_in_progress is False
@pytest.mark.parametrize(
"mutate",
[
lambda b: setattr(b, "_mtp_runtime_fallback_active", False),
lambda b: setattr(b, "_last_load_kwargs", None),
lambda b: setattr(b, "_process", None),
lambda b: b._cancel_event.set(),
],
)
def test_runtime_recovery_skips_when_not_applicable(monkeypatch, mutate):
# No reload when this launch is not running MTP+tensor, there is no snapshot,
# the process handle is gone, or the request was cancelled.
b = _recovery_backend()
mutate(b)
calls = []
monkeypatch.setattr(b, "load_model", lambda **k: calls.append(k))
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is False
assert calls == []
class _BlockingDeadProc:
# Reports alive until released, then dead -- lets a test mutate backend state
# while the recovery thread is still in its death-confirm poll.
def __init__(self):
self._dead = threading.Event()
def poll(self):
return 0 if self._dead.is_set() else None
def terminate(self):
self._dead.set()
def kill(self):
self._dead.set()
def wait(self, timeout = None):
self._dead.set()
return 0
def release(self):
self._dead.set()
def test_runtime_recovery_fires_for_user_env_mtp(monkeypatch):
# MTP driven by user extra_args / LLAMA_ARG_SPEC_TYPE leaves _speculative_type
# unset, but the launch flag still gates recovery on (pass-through MTP).
b = _recovery_backend()
b._speculative_type = None # Studio stepped back; user/env owns the spec
done = threading.Event()
captured = {}
def _fake_load_model(**kwargs):
captured.update(kwargs)
done.set()
return True
monkeypatch.setattr(b, "load_model", _fake_load_model)
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True
assert done.wait(timeout = 5)
assert captured["speculative_type"] == "off"
def test_runtime_recovery_strips_user_mtp_extra_args(monkeypatch):
# A user --spec-type draft-mtp in extra_args must be neutralised on the reload
# (append a last-wins --spec-default) so MTP can't re-engage and loop.
b = _recovery_backend()
b._last_load_kwargs = dict(b._last_load_kwargs, extra_args = ["--spec-type", "draft-mtp"])
done = threading.Event()
captured = {}
def _fake_load_model(**kwargs):
captured.update(kwargs)
done.set()
return True
monkeypatch.setattr(b, "load_model", _fake_load_model)
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True
assert done.wait(timeout = 5)
assert captured["speculative_type"] == "off"
assert captured["extra_args"][-1] == "--spec-default"
def test_runtime_recovery_restores_requested_mode(monkeypatch):
# After the off-reload, /status must show the user's requested mode + the
# runtime-error note, not a bare "off" (matches the startup MTP fallback).
b = _recovery_backend()
b._last_load_kwargs = dict(b._last_load_kwargs, speculative_type = "mtp")
done = threading.Event()
def _fake_load_model(**kwargs):
b._requested_spec_mode = "off" # what a real off-reload would leave behind
done.set()
return True
monkeypatch.setattr(b, "load_model", _fake_load_model)
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True
assert done.wait(timeout = 5)
deadline = time.monotonic() + 2
while b._requested_spec_mode != "mtp" and time.monotonic() < deadline:
time.sleep(0.02)
assert b._requested_spec_mode == "mtp"
assert b._spec_fallback_reason == "runtime_error"
def test_runtime_recovery_skips_when_process_replaced(monkeypatch):
# A newer user load that replaces the process during the death-confirm poll
# must not be clobbered by the stale recovery replay.
b = _recovery_backend()
p1 = _BlockingDeadProc()
b._process = p1
calls = []
monkeypatch.setattr(b, "load_model", lambda **k: calls.append(k))
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True # captures p1
b._process = _FakeProcess() # a newer load swapped the live process
p1.release() # p1 now reports dead -> recovery runs its staleness check
time.sleep(0.6)
assert calls == [], "stale recovery replayed over a newer load"
def test_runtime_recovery_skips_when_snapshot_changed(monkeypatch):
# If the recorded load changed during the poll, the stale snapshot is dropped.
b = _recovery_backend()
p1 = _BlockingDeadProc()
b._process = p1
calls = []
monkeypatch.setattr(b, "load_model", lambda **k: calls.append(k))
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True
b._last_load_kwargs = dict(b._last_load_kwargs, model_identifier = "other/model")
p1.release()
time.sleep(0.6)
assert calls == []
def test_runtime_recovery_is_single_flight(monkeypatch):
# Concurrent failures schedule only one reload.
b = _recovery_backend()
started = threading.Event()
release = threading.Event()
def _slow_load(**kwargs):
started.set()
release.wait(timeout = 5)
return True
monkeypatch.setattr(b, "load_model", _slow_load)
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is True
assert started.wait(timeout = 5)
# Second failure while the first reload is in flight is a no-op.
assert b._maybe_recover_from_mtp_crash(RuntimeError()) is False
release.set()
def test_runtime_recovery_rechecks_cancel_before_reload():
# recover() must re-check the cancel flag after the death poll (load_model
# clears it), so a reload scheduled just before /unload can't resurrect it.
src = inspect.getsource(LlamaCppBackend._maybe_recover_from_mtp_crash)
cancel = src.rfind("self._cancel_event.is_set()")
load = src.find("self.load_model(")
assert 0 <= cancel < load, "recovery must re-check cancel before reloading"
def test_probe_mtp_decode_uses_api_key_auth(monkeypatch):
# Direct-stream mode runs llama-server with --api-key; the probe must send
# the same bearer auth or it gets a spurious 401 and falsely drops MTP.
backend = LlamaCppBackend()
backend._port = 0
backend._process = None
captured = {}
class _Resp:
status_code = 200
def _capture(*a, **k):
captured.clear()
captured.update(k)
return _Resp()
monkeypatch.setattr(llama_cpp_module.httpx, "post", _capture, raising = False)
backend._api_key = "secret"
backend._probe_mtp_decode(timeout = 1.0)
assert captured["headers"] == {"Authorization": "Bearer secret"}
assert captured["trust_env"] is False
backend._api_key = None
backend._probe_mtp_decode(timeout = 1.0)
assert captured["headers"] is None
class _ToggleProcess:
"""A subprocess stand-in whose liveness can be flipped at runtime."""
def __init__(self):
self._alive = True
def poll(self):
return None if self._alive else 0
def terminate(self):
self._alive = False
def kill(self):
self._alive = False
def wait(self, timeout = None):
self._alive = False
return 0
def die(self):
self._alive = False
def test_crash_watchdog_triggers_recovery_on_death(monkeypatch):
# The watchdog must notice the process exit and recover even when no request
# handler observed it (e.g. the direct proxy endpoints).
b = _recovery_backend()
proc = _ToggleProcess()
b._process = proc
fired = threading.Event()
monkeypatch.setattr(b, "_maybe_recover_from_mtp_crash", lambda *a, **k: fired.set())
b._start_mtp_crash_watchdog()
assert b._mtp_watchdog_thread is not None
proc.die()
assert fired.wait(timeout = 3)
def test_crash_watchdog_ignores_intentional_termination(monkeypatch):
# A planned reload/unload stops the watchdog before killing the process, so
# the resulting death must not be mistaken for a crash.
b = _recovery_backend()
proc = _ToggleProcess()
b._process = proc
fired = threading.Event()
monkeypatch.setattr(b, "_maybe_recover_from_mtp_crash", lambda *a, **k: fired.set())
b._start_mtp_crash_watchdog()
b._stop_mtp_crash_watchdog() # what _kill_process does first
proc.die()
assert not fired.wait(timeout = 2)
assert b._mtp_watchdog_thread is None
@pytest.mark.parametrize(
"mutate",
[
lambda b: setattr(b, "_mtp_runtime_fallback_active", False),
lambda b: setattr(b, "_process", None),
],
)
def test_crash_watchdog_not_armed_when_inapplicable(mutate):
# Only a launch actually running MTP+tensor with a live process arms it.
b = _recovery_backend()
b._process = _ToggleProcess()
mutate(b)
b._start_mtp_crash_watchdog()
assert b._mtp_watchdog_thread is None
def test_kill_process_stops_crash_watchdog(monkeypatch):
# _kill_process is the single deliberate-termination chokepoint; it must
# stop the watchdog so the planned kill isn't seen as a crash.
b = _recovery_backend()
proc = _ToggleProcess()
b._process = proc
fired = threading.Event()
monkeypatch.setattr(b, "_maybe_recover_from_mtp_crash", lambda *a, **k: fired.set())
b._start_mtp_crash_watchdog()
b._kill_process()
assert b._mtp_watchdog_thread is None
assert b._process is None
assert not fired.wait(timeout = 2)
def test_kill_process_stops_watchdog_before_terminate():
# Ordering matters: stop the watchdog before terminating so the watchdog's
# post-death stop re-check reliably sees a planned kill.
src = inspect.getsource(LlamaCppBackend._kill_process)
stop = src.find("_stop_mtp_crash_watchdog()")
term = src.find(".terminate(")
assert 0 <= stop < term, "must stop the watchdog before terminating"
def test_crash_watchdog_rechecks_stop_before_recovery():
# After a detected exit the watchdog re-checks the stop flag so a kill that
# raced in between the poll-wait and the poll-read can't fire recovery.
src = inspect.getsource(LlamaCppBackend._start_mtp_crash_watchdog)
check = src.find("stop.is_set()")
recover = src.find("_maybe_recover_from_mtp_crash")
assert 0 <= check < recover, "must re-check stop before recovering"
def test_load_model_arms_crash_watchdog():
# The healthy-load commit arms the watchdog for this load.
src = inspect.getsource(LlamaCppBackend.load_model)
assert "_start_mtp_crash_watchdog" in src
# ── tensor-mode allocation: conservative VRAM budget ─────────────────
def _kv_seeded_backend() -> LlamaCppBackend:
# Minimal GGUF metadata so _can_estimate_kv() is True (legacy KV path).
backend = LlamaCppBackend()
backend._n_layers = 32
backend._embedding_length = 4096
backend._n_heads = 32
backend._n_kv_heads = 8
backend._context_length = 131072
return backend
def test_fit_context_budget_frac_override_is_tighter():
backend = _kv_seeded_backend()
model_size = 8 * 1024**3
pool_mib = 24 * 1024 # tight enough that KV capping bites
fit_default = backend._fit_context_to_vram(131072, pool_mib, model_size, "f16")
fit_tp = backend._fit_context_to_vram(131072, pool_mib, model_size, "f16", budget_frac = 0.80)
assert fit_tp < 131072, "expected the context to be capped at this VRAM tier"
assert fit_tp <= fit_default, "a tighter budget must not allow MORE context"
# Omitting the override must reproduce the default budget exactly.
assert backend._fit_context_to_vram(131072, pool_mib, model_size, "f16") == fit_default
# ── unsupported-arch load failure -> clean message ───────────────────
def test_split_mode_tensor_arch_failure_message():
msg = LlamaCppBackend._classify_llama_start_failure(
"llama_model_create: LLAMA_SPLIT_MODE_TENSOR not implemented for "
"architecture 'deepseek2'",
None,
"unsloth/DeepSeek-V3-GGUF",
)
assert "Tensor parallelism is not supported" in msg
def test_unrelated_arch_failure_not_hijacked_by_tensor_message():
msg = LlamaCppBackend._classify_llama_start_failure(
"unknown model architecture: 'flux'", "/models/flux.gguf", None
)
assert "Tensor parallelism" not in msg
# ── _plan_tensor_parallel: the allocation math (pure, no model/GPU) ───
# Seeded full-attention KV (~128 KiB/token) via _kv_seeded_backend, so the
# context cap + split are deterministic. Asserts relationships rather than
# magic numbers so the KV estimate can evolve without breaking these.
_GB = 1024**3
_ASYM = [(0, 48000), (1, 24000)] # asymmetric pool, 72000 MiB
_SYM = [(0, 24000), (1, 24000)] # symmetric pool
def _plan(
model_gb,
target = 131072,
gpus = _ASYM,
mtp = False,
):
b = _kv_seeded_backend()
return b, b._plan_tensor_parallel(gpus, int(model_gb * _GB), target, mtp_engaged = mtp)
def _kv_budget_b(model_gb, gpus = _ASYM):
# No totals here, so usable is the legacy free*frac (keeps the 5% cushion).
reserve = LlamaCppBackend._TENSOR_PARALLEL_BUFFER_RESERVE_MIB
usable = sum(f * _CTX_FIT_VRAM_FRACTION for _, f in gpus)
return (usable - len(gpus) * reserve) * 1024 * 1024 - int(model_gb * _GB)
def test_tp_plan_weighted_split_on_asymmetric_big_model():
b, (ec, mac, gi, ts) = _plan(50)
reserve = b._TENSOR_PARALLEL_BUFFER_RESERVE_MIB
assert gi == [0, 1]
# split weighted by (usable - flat buffer - per-device context compute); with
# no totals usable is free*frac. The per-device cc is subtracted so the smaller
# card isn't weighted above its real usable budget (see below).
cc_per_dev = b._compute_buffer_ctx_bytes(ec, None, None) // (1024 * 1024)
assert cc_per_dev > 0
assert ts == [
int(48000 * _CTX_FIT_VRAM_FRACTION - reserve - cc_per_dev),
int(24000 * _CTX_FIT_VRAM_FRACTION - reserve - cc_per_dev),
]
assert ec < 131072 # capped below native
def test_tp_plan_even_split_when_model_fits():
# A small model whose even share fits the smallest GPU -> llama.cpp's even
# default (None), which is safe for archs that crash on a weighted split.
_, (ec, mac, gi, ts) = _plan(4)
assert ts is None
def test_tp_plan_symmetric_gpus_use_even_split():
_, (ec, mac, gi, ts) = _plan(8, gpus = _SYM)
assert ts is None
def test_tp_plan_context_fits_pool_budget_no_oom():
b, (ec, mac, gi, ts) = _plan(50)
# the chosen context's KV must fit the pooled budget (weights + buffers)
assert b._estimate_kv_cache_bytes(ec) <= _kv_budget_b(50)
def test_tp_plan_uses_available_vram_not_wasteful():
# when the cap engages, the chosen context nearly fills the budget
b, (ec, mac, gi, ts) = _plan(50)
assert b._estimate_kv_cache_bytes(ec) >= 0.9 * _kv_budget_b(50)
def test_tp_plan_weights_exceed_pool_floors_context():
# 70 GB > pool minus per-GPU reserves -> floor (triggers layer fallback)
_, (ec, mac, gi, ts) = _plan(70)
assert ec == 2048
def test_tp_plan_floor_never_exceeds_explicit_small_context():
# An explicit context below the 2048 floor must not be raised: a caller
# asking for 1024 should not have KV sized for 2048 (avoidable OOM).
_, (ec, mac, gi, ts) = _plan(70, target = 1024) # weights exceed pool -> floor path
assert ec == 1024
_, (ec2, *_rest) = _plan(50, target = 1024) # cap path with a tiny budget
assert ec2 <= 1024
def test_tp_plan_explicit_context_honored_when_it_fits():
_, (ec, mac, gi, ts) = _plan(50, target = 8192)
assert ec == 8192
def test_tp_plan_explicit_context_capped_when_too_large():
_, (ec, mac, gi, ts) = _plan(50, target = 131072)
assert 2048 <= ec < 131072
def test_tp_plan_max_available_ctx_reports_native_not_explicit_ctx():
# An explicit small ctx caps effective_ctx but the UI ceiling
# (max_available_ctx) must reflect the native/hardware cap, not the request.
b = _kv_seeded_backend()
ec, mac, _gi, _ts = b._plan_tensor_parallel(_ASYM, int(50 * _GB), 8192, max_target_ctx = 131072)
_, native_mac, *_ = b._plan_tensor_parallel(_ASYM, int(50 * _GB), 131072)
assert ec == 8192 # explicit request honored for the load
assert mac == native_mac > ec # ceiling reflects the hardware cap
def test_tp_plan_mtp_reserves_extra_and_shrinks_context():
_, (ec_no, *_rest) = _plan(50)
_, (ec_mtp, *_rest) = _plan(50, mtp = True)
assert ec_mtp < ec_no
def test_tp_plan_reserves_context_linear_compute_buffer():
# Tensor mode replicates the compute graph on every device; measured on
# Qwen3.5-9B at f16 the per-device buffer grows ~n_ubatch*2 B/token (~1024
# B/tok), so the fit must reserve n_dev x that on top of the flat reserve or
# it over-pins and OOMs at high context. The chosen KV must leave room for it.
b, (ec, mac, gi, ts) = _plan(50)
cc = len(gi) * b._compute_buffer_ctx_bytes(ec, None, "f16")
assert cc > 0
assert b._estimate_kv_cache_bytes(ec) + cc <= _kv_budget_b(50)
def test_tp_plan_context_shrinks_vs_compute_unaware():
# With the context-linear term the pinned context is strictly below what a
# KV-only (compute-unaware) fit at the same budget would allow.
b, (ec, *_r) = _plan(50)
b2 = _kv_seeded_backend()
b2._embedding_length = 0 # kills the context-linear compute term (returns 0)
ec_naive, *_r2 = b2._plan_tensor_parallel(_ASYM, int(50 * _GB), 131072)
assert ec < ec_naive
def test_tp_plan_soft_overhead_shrinks_context():
# The CUDA-ctx / mmproj / MTP-draft reserve the layer path folds into the fit
# budget (model_size_fit) must also shrink the tensor context. Tensor mode has
# no --fit valve, so an unreserved overshoot OOMs at startup instead of
# offloading. A non-zero soft_overhead must pin a strictly smaller context.
b = _kv_seeded_backend()
ec_no, *_r = b._plan_tensor_parallel(_ASYM, int(50 * _GB), 131072)
ec_soft, *_r2 = b._plan_tensor_parallel(
_ASYM, int(50 * _GB), 131072, soft_overhead_bytes = 2 * _GB
)
assert 2048 < ec_soft < ec_no
def test_tp_plan_soft_overhead_reserved_against_budget():
# The pinned context must leave the whole soft reserve free on top of KV and
# the replicated context compute, so the real footprint stays within the pool.
b = _kv_seeded_backend()
soft = 2 * _GB
ec, *_r = b._plan_tensor_parallel(_ASYM, int(50 * _GB), 131072, soft_overhead_bytes = soft)
cc = len(_ASYM) * b._compute_buffer_ctx_bytes(ec, None, None)
assert b._estimate_kv_cache_bytes(ec) + cc + soft <= _kv_budget_b(50)
def test_tp_plan_weighted_split_keeps_small_gpu_within_budget():
# Regression: the weighted split must subtract each device's replicated context
# compute (cc_bytes/n_dev), not just the flat reserve. Otherwise the smaller
# card is weighted above its usable budget and OOMs at launch. Model the split:
# llama.cpp distributes weights+KV by the tensor-split weights; every device
# also holds the flat reserve plus its per-device context compute.
b, (ec, mac, gi, ts) = _plan(50)
assert ts is not None and len(ts) == len(gi) == 2
reserve = b._TENSOR_PARALLEL_BUFFER_RESERVE_MIB
cc_per_dev = b._compute_buffer_ctx_bytes(ec, None, None) // (1024 * 1024)
free_by_idx = {0: 48000, 1: 24000}
split_content_mib = (int(50 * _GB) + b._estimate_kv_cache_bytes(ec)) / (1024 * 1024)
total_weight = sum(ts)
for w, idx in zip(ts, gi):
placed = split_content_mib * w / total_weight
usable = free_by_idx[idx] * _CTX_FIT_VRAM_FRACTION
assert placed + reserve + cc_per_dev <= usable + 1 # +1 MiB for int rounding
# Lock the regression: under the old formula (flat reserve only) the smaller
# card was placed over its budget; the cc term is what pulls it back.
old_adj = [int(free_by_idx[i] * _CTX_FIT_VRAM_FRACTION - reserve) for i in gi]
old_small_placed = split_content_mib * old_adj[1] / sum(old_adj)
assert old_small_placed + reserve + cc_per_dev > free_by_idx[1] * _CTX_FIT_VRAM_FRACTION
def test_tp_plan_no_kv_metadata_floors_context():
b = LlamaCppBackend() # no KV metadata -> can't size safely
ec, mac, gi, ts = b._plan_tensor_parallel(_ASYM, int(50 * _GB), 131072)
assert ec <= 4096
def test_tp_plan_single_gpu_never_splits():
# The toggle is a no-op without >= 2 GPUs (most dev/CI machines). Even if
# the planner is reached, it must not emit a tensor split.
b = _kv_seeded_backend()
ec, mac, gi, ts = b._plan_tensor_parallel([(0, 24000)], int(8 * _GB), 8192)
assert ts is None
assert gi == [0]
def test_tp_plan_zero_gpus_never_splits():
b = _kv_seeded_backend()
ec, mac, gi, ts = b._plan_tensor_parallel([], int(8 * _GB), 8192)
assert ts is None
assert gi == []
def test_tp_plan_drops_gpu_below_buffer_reserve():
# A GPU with less free VRAM than the per-device compute-buffer reserve
# can't host tensor mode; it's excluded, which here leaves <2 usable -> no
# split (and gpu_indices reflects only the usable device).
b = _kv_seeded_backend()
reserve = LlamaCppBackend._TENSOR_PARALLEL_BUFFER_RESERVE_MIB
ec, mac, gi, ts = b._plan_tensor_parallel([(0, 48000), (1, reserve - 1)], int(8 * _GB), 8192)
assert gi == [0]
assert ts is None
# ── route auto-fallback survives a *raised* tensor-load crash ─────────
# A tensor-incompatible model makes load_model RAISE (not return False); the
# /load fallback must catch it and retry with layer split (stripping --split-mode
# so the retry can't relaunch tensor), while a non-tensor load propagates.
class _RecordingLoader:
"""Fake ``attempt_load``: crashes whenever tensor mode is effectively
engaged (via the bool or a ``--split-mode`` in extras), like a real
tensor-incompatible model; succeeds on layer split."""
def __init__(self):
self.calls: list[tuple] = []
async def __call__(self, tensor_parallel, extra_args):
self.calls.append((tensor_parallel, list(extra_args) if extra_args else extra_args))
if resolve_tensor_parallel(extra_args, tensor_parallel):
raise RuntimeError("llama-server failed to start")
return True
def test_tensor_fallback_retries_layer_on_crash():
loader = _RecordingLoader()
ok = asyncio.run(
load_with_tensor_fallback(loader, requested_tensor = True, extra_args = None, label = "m")
)
assert ok is True
# tensor first (crashes), then layer split.
assert [c[0] for c in loader.calls] == [True, False]
def test_tensor_fallback_no_retry_on_success():
calls: list[bool] = []
async def _ok(tensor_parallel, extra_args):
calls.append(tensor_parallel)
return True
ok = asyncio.run(
load_with_tensor_fallback(_ok, requested_tensor = True, extra_args = None, label = "m")
)
assert ok is True
assert calls == [True] # no fallback when the tensor load succeeds
def test_tensor_fallback_retries_when_tensor_returns_false():
# load_model can signal failure by *returning False* (not only by raising);
# that must trigger the layer-split retry just like a crash does.
calls: list[bool] = []
async def _false_on_tensor(tensor_parallel, extra_args):
calls.append(tensor_parallel)
return not resolve_tensor_parallel(extra_args, tensor_parallel)
ok = asyncio.run(
load_with_tensor_fallback(
_false_on_tensor, requested_tensor = True, extra_args = None, label = "m"
)
)
assert ok is True
assert calls == [True, False]
def test_tensor_fallback_returns_false_when_both_attempts_fail():
# Tensor fails and the layer retry also fails -> the helper returns False so
# the route raises its own HTTP 500 (it does not crash mid-flight).
calls: list[bool] = []
async def _always_false(tensor_parallel, extra_args):
calls.append(tensor_parallel)
return False
ok = asyncio.run(
load_with_tensor_fallback(_always_false, requested_tensor = True, extra_args = None, label = "m")
)
assert ok is False
assert calls == [True, False] # tried tensor, then layer split
def test_tensor_fallback_skips_layer_retry_when_cancelled():
# load_model returns False on a user cancellation too. When cancelled() is
# True, the helper must NOT relaunch the load the user just cancelled.
calls: list[bool] = []
async def _false_on_tensor(tensor_parallel, extra_args):
calls.append(tensor_parallel)
return False
ok = asyncio.run(
load_with_tensor_fallback(
_false_on_tensor,
requested_tensor = True,
extra_args = None,
label = "m",
cancelled = lambda: True,
)
)
assert ok is False
assert calls == [True] # no layer-split retry after cancellation
@pytest.mark.parametrize(
"extras",
[
["--split-mode", "tensor", "-c", "4096"],
["-sm", "tensor", "-c", "4096"],
["--split-mode=tensor", "-c", "4096"],
["-sm=tensor", "-c", "4096"],
],
)
def test_tensor_fallback_strips_split_mode_from_extras_on_retry(extras):
# Tensor engaged via extras (boolean False); the retry must drop every
# --split-mode form (long/short, space/=) and force layer, keeping the user's
# other flags, else tensor is re-enabled and relaunches the crash.
loader = _RecordingLoader()
ok = asyncio.run(
load_with_tensor_fallback(loader, requested_tensor = False, extra_args = extras, label = "m")
)
assert ok is True
assert len(loader.calls) == 2
# User --split-mode replaced by an explicit layer override; -c kept.
assert loader.calls[1][1] == ["-c", "4096", "--split-mode", "layer"]
def test_tensor_fallback_env_tensor_retry_forces_layer(monkeypatch):
# Env-only tensor (toggle off, no --split-mode extra): load_model engages
# tensor via LLAMA_ARG_SPLIT_MODE and a tensor-incompatible model crashes. The
# wrapper must (1) recognise the env tensor request and retry, and (2) force
# --split-mode layer so the retry doesn't re-engage tensor via the still-set
# env and crash again (#6312).
monkeypatch.setenv("LLAMA_ARG_SPLIT_MODE", "tensor")
calls: list = []
async def _crash_when_effectively_tensor(tensor_parallel, extra_args):
calls.append(list(extra_args) if extra_args else extra_args)
# Mirror real load_model: env-aware tensor engagement crashes.
if _effective_tensor_parallel(extra_args, tensor_parallel):
raise RuntimeError("llama-server failed to start (tensor)")
return True
ok = asyncio.run(
load_with_tensor_fallback(
_crash_when_effectively_tensor,
requested_tensor = False,
extra_args = None,
label = "m",
)
)
assert ok is True
assert len(calls) == 2
# The forced layer override neutralises the inherited tensor env on retry.
assert calls[1] == ["--split-mode", "layer"]
def test_tensor_fallback_propagates_non_tensor_crash():
async def _always_raise(tensor_parallel, extra_args):
raise RuntimeError("bad model")
with pytest.raises(RuntimeError, match = "bad model"):
asyncio.run(
load_with_tensor_fallback(
_always_raise, requested_tensor = False, extra_args = None, label = "m"
)
)
# ── _plan_tensor_parallel: total-based headroom + ubatch (review fixes) ──
def test_tensor_caps_context_to_total_vram_budget():
# Partly-used 80 GB cards: 20 GB free each. With total_by_idx the planner must
# cap occupancy at 0.95*total (not spend the cushion the layer-split paths keep).
b = _kv_seeded_backend()
gpus = [(0, 20000), (1, 20000)]
totals = {0: 81920, 1: 81920}
model = int(18 * _GB)
with_total, *_ = b._plan_tensor_parallel(gpus, model, 131072, total_by_idx = totals)
without, *_ = b._plan_tensor_parallel(gpus, model, 131072)
assert with_total < without # total cap tightens the chosen context
MIB = 1024 * 1024
reserve = LlamaCppBackend._TENSOR_PARALLEL_BUFFER_RESERVE_MIB # flat (no vocab dims)
pool_usable = sum(f - (1.0 - _CTX_FIT_VRAM_FRACTION) * totals[i] for i, f in gpus)
foot_total = (model + b._estimate_kv_cache_bytes(with_total, None)) / MIB + len(gpus) * reserve
foot_free = (model + b._estimate_kv_cache_bytes(without, None)) / MIB + len(gpus) * reserve
assert foot_total <= pool_usable + 2 # fix: fits the total-based budget
assert foot_free > pool_usable # old behavior over-spent the cushion
def test_tensor_unknown_total_keeps_fraction_cushion():
# A two-column nvidia-smi probe yields total 0. The planner must fall back to
# free*frac (keep the 5% cushion), like _select_gpus/_gpu_usable, not raw free,
# or it over-advertises context exactly where the PR is hardening the budget.
b = _kv_seeded_backend()
gpus = [(0, 20000), (1, 20000)]
MIB = 1024 * 1024
reserve = LlamaCppBackend._TENSOR_PARALLEL_BUFFER_RESERVE_MIB
model = int(18 * _GB)
ec_zero, *_ = b._plan_tensor_parallel(gpus, model, 131072, total_by_idx = {0: 0, 1: 0})
ec_none, *_ = b._plan_tensor_parallel(gpus, model, 131072)
assert ec_zero == ec_none # total 0 == total absent: both use free*frac
pool_free = sum(f for _, f in gpus)
foot = (model + b._estimate_kv_cache_bytes(ec_zero, None)) / MIB + len(gpus) * reserve
assert foot <= pool_free * _CTX_FIT_VRAM_FRACTION + 2 # within free*frac, not raw free
def test_tensor_reserve_scales_with_ubatch():
# A user --ubatch override must enlarge the per-device reserve -> less ctx room.
b = _kv_seeded_backend()
b._vocab_size = 152064 # enable the deterministic compute-buffer estimate
gpus = [(0, 16000), (1, 16000)]
model = int(18 * _GB)
small_ub, *_ = b._plan_tensor_parallel(gpus, model, 131072, n_ubatch = 512)
big_ub, *_ = b._plan_tensor_parallel(gpus, model, 131072, n_ubatch = 4096)
assert big_ub < small_ub
def test_plan_tensor_carries_unsized_mtp_flat_reserve():
# review run3 #1/#5: with a weights-only (KV-unsized) MTP reserve, the planner
# gets a non-None mtp_overhead_fn but must still subtract the flat unsized-KV
# cushion, or its binary search spends it on context. Passing the reserve must
# pick a strictly smaller context than passing 0.
b = _kv_seeded_backend()
gpus = [(0, 14000), (1, 14000)] # tight pool so the context is actually capped
model = int(8 * _GB)
weights_only = lambda c: 3 * _GB # noqa: E731 -- constant drafter weights, no KV term
ctx_no_flat, *_ = b._plan_tensor_parallel(
gpus,
model,
131072,
mtp_engaged = True,
mtp_overhead_fn = weights_only,
mtp_flat_reserve_bytes = 0,
)
ctx_flat, *_ = b._plan_tensor_parallel(
gpus,
model,
131072,
mtp_engaged = True,
mtp_overhead_fn = weights_only,
mtp_flat_reserve_bytes = 2 * _GB,
)
assert 0 < ctx_flat < ctx_no_flat
def test_tensor_admission_drops_gpu_below_usable_budget():
# A partly-used big card can clear the buffer reserve on raw free yet have no
# usable budget left (free - 0.05*total). Admit by usable budget: GPU 0 here is
# 6000 free on an 80 GB card -> usable 1904 < flat reserve 5120, so it's dropped
# (leaving <2 -> no split). Without total_by_idx, raw free 6000 >= 5120 admits it.
b = _kv_seeded_backend()
gpus = [(0, 6000), (1, 40000)]
totals = {0: 81920, 1: 81920}
_ec, _mac, gi, ts = b._plan_tensor_parallel(gpus, int(8 * _GB), 8192, total_by_idx = totals)
assert gi == [1] and ts is None # GPU 0 excluded on usable budget
_ec2, _mac2, gi_raw, _ts2 = b._plan_tensor_parallel(gpus, int(8 * _GB), 8192)
assert gi_raw == [0, 1] # raw free would have admitted both
def test_load_model_tensor_admission_and_capacity_gate_use_usable_budget():
# load_model is too entangled (subprocess + GPU probe) to drive end-to-end, so
# assert at the source level that the tensor prefilter admits on the usable
# budget (_gpu_usable), not raw free, and downgrades to layer split when the
# pooled budget can't hold weights + per-device compute buffers.
src = inspect.getsource(LlamaCppBackend.load_model)
assert "_gpu_usable(g) >= reserve_mib" in src # admit by usable budget
assert "g[1] >= reserve_mib" not in src # not raw free
assert "_tp_weight_budget_mib" in src # pooled-weight capacity gate
assert "falling back to layer split" in src # downgrade on overcommit
# The gate's required footprint must include the non-shrinkable MTP reserve,
# not weights alone, or a separate-drafter MTP load can still overcommit.
assert "_tp_mtp_floor" in src
assert "model_size + _tp_mtp_floor" in src
def test_load_model_tensor_floor_keeps_flat_reserve_for_weights_only():
# Tensor mode has no --fit valve, so a weights-only drafter (KV unsized) must
# keep the flat reserve as the draft-KV cushion, not just the byte weights
# (Finding H1, the tensor analog of the layer-split _mtp_kv_unsized handling).
compact = "".join(inspect.getsource(LlamaCppBackend.load_model).split())
# byte-only floor used only when KV is sizable (not the weights-only case)
assert "mtp_overhead_fnisnotNoneandnot_mtp_kv_unsized" in compact
# weights-only / dims-unavailable: flat reserve, never below the byte floor
assert "_tp_mtp_floor=max(" in compact
def test_load_model_reserves_pipeline_per_device_overhead():
# Layer split must reserve the fixed per-device overhead per EXTRA device so a
# tight multi-GPU split can't pin a context that OOMs a device (Finding A); k=1
# adds nothing.
assert LlamaCppBackend._PIPELINE_PER_DEVICE_OVERHEAD_MIB > 0
compact = "".join(inspect.getsource(LlamaCppBackend.load_model).split())
assert "def_subset_model_size(n_gpus:int)->int:" in compact
assert "max(0,n_gpus-1)*_pipeline_overhead_bytes" in compact
assert "_subset_model_size(n_gpus)" in compact # used in the layer-split fit
def test_load_model_restores_quantized_kv_on_tensor_downgrade():
# A quantized KV dropped for the tensor attempt must be restored if tensor
# downgrades to layer split (Finding D); captured once, restored at both the
# GPU-count and capacity-gate downgrades.
compact = "".join(inspect.getsource(LlamaCppBackend.load_model).split())
assert "_tensor_dropped_cache_type_kv=cache_type_kv" in compact # captured pre-null
# Restore is shared in one closure, called at every tensor->layer downgrade.
assert "cache_type_kv=_tensor_dropped_cache_type_kv" in compact # restored in the closure
assert "def_restore_after_tensor_downgrade():" in compact # one shared restore helper
assert compact.count("_restore_after_tensor_downgrade()") >= 3 # called at each downgrade