fitcheck / tests /test_display_render.py
mlanvvs's picture
Deploy FitCheck Gradio demo
df31653 verified
Raw History Blame Contribute Delete
18.9 kB
"""What the rich renderables actually say.
Every panel is rendered through a console of fixed width, so an assertion is about
the words on the screen and not about this terminal's size. The estimator is not
under test here -- several reports are hand-built so that a branch can be reached
without hunting for a config that happens to trigger it.
"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import replace
from typing import Any
import pytest
from fitcheck.advisor import SweepSpec, advise
from fitcheck.config_parser import ModelConfig, fetch_model_config
from fitcheck.display import (
_activation_label,
_adapter_label,
_attention_label,
_batch_label,
_hint_savings_mib,
_lora_label,
_optimizer_label,
_params_label,
_weights_label,
render_advisor_report,
render_explanation,
render_gpu_table,
render_inference_report,
render_report,
render_verbose_detail,
use_ascii_glyphs,
)
from fitcheck.estimator import (
InferenceReport,
MemoryReport,
ServingConfig,
TrainingConfig,
estimate,
estimate_inference,
)
from fitcheck.gpu_db import GpuSpec, get_gpu
from rich.console import Console
def _text(renderable: object, width: int = 200) -> str:
console = Console(width=width, color_system=None, legacy_windows=False)
with console.capture() as capture:
console.print(renderable)
return " ".join(capture.get().split())
@pytest.fixture
def llama_model(
fake_config_download: Callable[[dict[str, Any]], None],
llama_31_8b_config: dict[str, Any],
) -> ModelConfig:
fake_config_download(llama_31_8b_config)
return fetch_model_config("meta-llama/Llama-3.1-8B")
@pytest.fixture
def small_model(
fake_config_download: Callable[[dict[str, Any]], None],
mha_config: dict[str, Any],
) -> ModelConfig:
fake_config_download(mha_config)
return fetch_model_config("openai-community/gpt2")
@pytest.fixture
def mqa_model(
fake_config_download: Callable[[dict[str, Any]], None],
mqa_config: dict[str, Any],
) -> ModelConfig:
fake_config_download(mqa_config)
return fetch_model_config("tiny/mqa")
@pytest.fixture
def qlora_training() -> TrainingConfig:
return TrainingConfig(
precision="bf16",
quantization="nf4",
optimizer="adamw",
optimizer_dtype="fp32",
batch_size=1,
seq_len=2048,
lora_rank=64,
lora_targets=["q_proj", "k_proj", "v_proj", "o_proj"],
grad_checkpoint=True,
flash_attn=True,
)
def _report(**overrides: Any) -> MemoryReport:
base: dict[str, Any] = {
"weight_mib": 1_000.0,
"lora_mib": 10.0,
"optimizer_mib": 20.0,
"gradient_mib": 10.0,
"activation_mib": 400.0,
"overhead_mib": 560.0,
"total_mib": 2_000.0,
"gpu_capacity_mib": 4_000.0,
"headroom_mib": 2_000.0,
"fits": True,
"max_batch_size": 4,
"effective_batch_size": 1,
"savings_hints": [],
"warnings": (),
}
base.update(overrides)
return MemoryReport(**base)
def test_attention_label_names_each_family(
llama_model: ModelConfig, small_model: ModelConfig, mqa_model: ModelConfig
) -> None:
assert _attention_label(small_model) == "12 heads (MHA)"
assert _attention_label(mqa_model) == "16 heads (MQA)"
assert _attention_label(llama_model) == "32 heads, GQA 8 KV heads"
def test_params_label_switches_unit_below_a_billion(
llama_model: ModelConfig, small_model: ModelConfig
) -> None:
assert _params_label(llama_model.num_params) == "8.03B params"
assert _params_label(small_model.num_params).endswith("M params")
def test_adapter_and_component_labels_follow_the_config(
qlora_training: TrainingConfig,
) -> None:
lora = replace(qlora_training, quantization="none")
full = replace(qlora_training, lora_rank=None)
assert _adapter_label(qlora_training).startswith("QLoRA r=64 [q,k,v,o]")
assert _adapter_label(lora).startswith("LoRA r=64")
assert _adapter_label(full) == "full fine-tune"
assert _lora_label(full) == "LoRA adapter (none, full fine-tune)"
assert _weights_label(lora) == "Base model weights (bf16)"
assert _weights_label(replace(qlora_training, double_quant=True)) == (
"Base model weights (NF4 + double quant)"
)
def test_labels_degrade_gracefully_without_a_training_config() -> None:
assert _weights_label(None) == "Base model weights"
assert _lora_label(None) == "LoRA adapter (trainable)"
assert _activation_label(None) == "Activations"
def test_activation_and_batch_and_optimizer_labels(
qlora_training: TrainingConfig,
) -> None:
plain = replace(qlora_training, grad_checkpoint=False, flash_attn=False)
assert _activation_label(plain) == "Activations (no ckpt, no flash)"
assert _batch_label(replace(qlora_training, batch_size=2, grad_accum_steps=8)) == (
"bs 2 x 8 accum = 16 effective"
)
assert _optimizer_label(qlora_training) == "adamw (fp32 states)"
assert _optimizer_label(replace(qlora_training, optimizer="sgd")) == "sgd"
def test_report_renders_without_a_training_config(
llama_model: ModelConfig,
) -> None:
output = _text(render_report(_report(), llama_model, get_gpu("4090")))
assert "Base model weights" in output
assert "Max micro-batch size at this sequence length: 4." in output
assert "Config" not in output
def test_report_flags_a_tight_fit(llama_model: ModelConfig) -> None:
tight = _report(total_mib=3_800.0, headroom_mib=200.0)
output = _text(render_report(tight, llama_model, get_gpu("4090"), ascii_only=True))
assert "FITS, BARELY" in output
assert "[!]" in output
def test_report_suggests_dropping_the_batch_when_it_does_not_fit(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, batch_size=8, grad_accum_steps=2)
over = _report(
total_mib=9_000.0, headroom_mib=-5_000.0, fits=False,
max_batch_size=2, effective_batch_size=16,
)
output = _text(render_report(over, llama_model, get_gpu("4090"), training))
assert "DOES NOT FIT" in output
assert "Drop batch_size to 2 to fit" in output
assert "keep the effective batch at 16" in output
def test_report_says_when_the_batch_is_already_at_the_ceiling(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, batch_size=4)
output = _text(
render_report(_report(max_batch_size=4), llama_model, get_gpu("4090"), training)
)
assert "batch_size 4 is already the maximum" in output
def test_report_rescues_a_config_that_does_not_fit_at_all(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
over = _report(fits=False, headroom_mib=-1.0, max_batch_size=0)
output = _text(render_report(over, llama_model, get_gpu("4090"), qlora_training))
assert "Even batch_size=1 does not fit" in output
def test_report_prints_warnings_and_the_best_savings_hint(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
report = _report(
savings_hints=[
"--flash-attn: saves 12 MiB",
"--quant nf4 -> saves 2,400 MiB",
],
warnings=("the MLP shape is assumed",),
)
output = _text(
render_report(report, llama_model, get_gpu("4090"), qlora_training, ascii_only=True)
)
assert "saves 2,400 MiB" in output
assert "the MLP shape is assumed" in output
def test_a_hint_with_no_parsable_saving_is_not_promoted(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
"""_SAVINGS_PATTERN misses: the hint grid must fall back, not crash."""
report = _report(savings_hints=["--flash-attn: saves nothing at this shape"])
output = _text(render_report(report, llama_model, get_gpu("4090"), qlora_training))
assert "saves nothing at this shape" not in output
assert _hint_savings_mib("--flash-attn: costs 5 MiB") == 0.0
assert _hint_savings_mib("saves 1,024 MiB") == 1_024.0
def test_a_zero_capacity_card_does_not_divide_by_zero(llama_model: ModelConfig) -> None:
empty = GpuSpec("Empty", 0, 0)
report = _report(gpu_capacity_mib=0.0, headroom_mib=0.0, fits=False)
assert "DOES NOT FIT" in _text(render_report(report, llama_model, empty))
@pytest.mark.parametrize(
("overrides", "expected"),
[
({"weight_mib": 9_000.0}, "the frozen base model at NF4"),
({"lora_mib": 9_000.0}, "rank 64 across 4 target modules"),
({"optimizer_mib": 9_000.0}, "AdamW keeps momentum and variance in fp32"),
({"gradient_mib": 9_000.0}, "one .grad tensor per trainable parameter"),
({"overhead_mib": 9_000.0}, "CUDA context"),
],
)
def test_explanation_explains_whichever_component_wins(
llama_model: ModelConfig,
qlora_training: TrainingConfig,
overrides: dict[str, Any],
expected: str,
) -> None:
report = _report(**overrides)
output = _text(
render_explanation(report, llama_model, qlora_training, get_gpu("4090"))
)
assert expected in output
def test_explanation_offers_nf4_when_the_base_is_unquantized(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, quantization="none")
output = _text(
render_explanation(_report(weight_mib=9_000.0), llama_model, training)
)
assert "Quantizing it with --quant nf4 cuts this to roughly a quarter" in output
def test_explanation_blames_the_checkpoint_store_when_the_layer_wins(
small_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
"""A 50k vocab and a long sequence put A_layer above A_logits."""
training = replace(
qlora_training, quantization="none", seq_len=8192, flash_attn=False
)
output = _text(
render_explanation(_report(activation_mib=9_000.0), small_model, training)
)
assert "checkpoint store plus one recomputed layer" in output
assert "Lowering --batch-size" in output
def test_explanation_points_at_grad_checkpointing_when_it_is_off(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, grad_checkpoint=False)
output = _text(
render_explanation(_report(activation_mib=9_000.0), llama_model, training)
)
assert "all 32 layers keep their full set of saved tensors" in output
assert "--grad-checkpoint trades compute" in output
def test_explanation_prices_a_non_adamw_optimizer(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, optimizer="adam8bit")
output = _text(
render_explanation(_report(optimizer_mib=9_000.0), llama_model, training)
)
assert "adam8bit states over the trainable parameters" in output
def test_explanation_falls_back_to_the_default_overhead_profile(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
"""A hump-form profile has no meaning without the checkpointed max()."""
training = replace(qlora_training, grad_checkpoint=False)
output = _text(
render_explanation(_report(overhead_mib=9_000.0), llama_model, training, get_gpu("t4"))
)
assert "CUDA context floor" in output
assert "of weights and activations" in output
def test_explanation_carries_the_warnings_through(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
report = _report(warnings=("int8 activations are modelled on one measured row",))
output = _text(render_explanation(report, llama_model, qlora_training))
assert "int8 activations are modelled on one measured row" in output
def test_verbose_detail_shows_the_checkpointed_arithmetic(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
output = _text(render_verbose_detail(llama_model, qlora_training))
assert "Checkpoint store" in output
assert "avoided by Flash Attention" in output
assert "never overlap" in output
assert "A_act charged" in output
def test_verbose_detail_shows_the_retained_form_without_checkpointing(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, grad_checkpoint=False, flash_attn=False)
output = _text(render_verbose_detail(llama_model, training))
assert "~3 copies retained per layer, no Flash Attention" in output
assert "x 32 layers, every layer's tensors kept" in output
assert "+ A_logits" in output
def test_verbose_detail_in_ascii_spells_the_symbols_out(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
unquantized = replace(qlora_training, quantization="none", flash_attn=False)
output = _text(render_verbose_detail(llama_model, unquantized, ascii_only=True))
assert "Checkpoint store (L x gamma*b*s*h)" in output
assert "included, no Flash Attention" in output
assert "A_logits, 3.5 fp32 copies of (b, s, V)" in output
assert "activation dtype" in output
def test_inference_report_names_the_cache_and_the_weights(
llama_model: ModelConfig,
) -> None:
serving = ServingConfig(quantization="nf4", seq_len=2048, num_concurrent=2)
report = estimate_inference(llama_model, serving, get_gpu("a100-80"))
output = _text(render_inference_report(report, llama_model, get_gpu("a100-80"), serving))
assert "2 concurrent requests" in output
assert "NF4 weights" in output
assert "KV cache (2 x 2,048 tokens, fp16)" in output
assert "MiB per token" in output
def test_inference_report_labels_an_unquantized_base(llama_model: ModelConfig) -> None:
serving = ServingConfig(num_concurrent=1)
report = estimate_inference(llama_model, serving, get_gpu("a100-80"))
output = _text(render_inference_report(report, llama_model, get_gpu("a100-80"), serving))
assert "1 request" in output
assert "unquantized weights" in output
assert "Base model weights (fp16)" in output
def _inference(**overrides: Any) -> InferenceReport:
base: dict[str, Any] = {
"weight_mib": 1_000.0,
"kv_cache_mib": 500.0,
"overhead_mib": 500.0,
"total_mib": 2_000.0,
"kv_mib_per_request": 500.0,
"kv_mib_per_token": 0.25,
"gpu_capacity_mib": 4_000.0,
"headroom_mib": 2_000.0,
"fits": True,
"max_concurrent": 4,
"warnings": (),
}
base.update(overrides)
return InferenceReport(**base)
@pytest.mark.parametrize(
("max_concurrent", "asked", "expected"),
[
(0, 1, "Not even one request fits"),
(4, 1, "Room for 4 concurrent requests"),
(2, 8, "Drop to 2 concurrent requests"),
(4, 4, "4 concurrent requests is already the maximum"),
],
)
def test_inference_concurrency_suggestions(
llama_model: ModelConfig, max_concurrent: int, asked: int, expected: str
) -> None:
serving = ServingConfig(num_concurrent=asked)
report = _inference(max_concurrent=max_concurrent, fits=max_concurrent > 0)
output = _text(
render_inference_report(report, llama_model, get_gpu("4090"), serving)
)
assert expected in output
def test_inference_report_prints_its_warnings(llama_model: ModelConfig) -> None:
report = _inference(warnings=("a paged engine allocates less",))
output = _text(
render_inference_report(
report, llama_model, get_gpu("4090"), ServingConfig(), ascii_only=True
)
)
assert "a paged engine allocates less" in output
def test_gpu_table_lists_every_card_with_its_usable_share() -> None:
output = _text(render_gpu_table())
assert "fitcheck GPU database" in output
assert "4090" in output
assert "t4" in output
assert "Usable %" in output
@pytest.fixture
def advisor_output(llama_model: ModelConfig, qlora_training: TrainingConfig) -> str:
sweep = SweepSpec(batch_sizes=(1, 2), seq_lens=(1024, 2048), lora_ranks=(8, 64))
report = advise(
llama_model,
qlora_training,
get_gpu("a100-80"),
sweep,
model_id="meta-llama/Llama-3.1-8B",
)
return _text(render_advisor_report(report, llama_model))
def test_advisor_report_draws_frontier_wall_and_prices(advisor_output: str) -> None:
assert "FRONTIER -- the edge of what fits" in advisor_output
assert "batch x seq splits at the same cost" in advisor_output
assert "Run it:" in advisor_output
assert "THE WALL -- exact ceilings, by bisection" in advisor_output
assert "Nearest 2^n" in advisor_output
assert "PRICE OF EACH AXIS -- from the same anchor" in advisor_output
def test_advisor_report_echoes_the_axes_it_held_and_swept(advisor_output: str) -> None:
assert "NF4 base" in advisor_output
assert "grad ckpt" in advisor_output
assert "flash attn" in advisor_output
assert "batch 1, 2" in advisor_output
assert "rank 8, 64" in advisor_output
def test_advisor_report_warns_that_tokens_per_step_is_not_a_speed(
advisor_output: str,
) -> None:
assert "not a speed" in advisor_output
assert "fitcheck has no throughput model" in advisor_output
def test_advisor_report_says_so_when_nothing_fits(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
sweep = SweepSpec(batch_sizes=(16,), seq_lens=(8192,), lora_ranks=(64,))
report = advise(llama_model, qlora_training, get_gpu("t4"), sweep)
output = _text(render_advisor_report(report, llama_model, ascii_only=True))
assert "nothing here fits" in output
assert "Nothing fits: add --grad-checkpoint" in output
assert "FRONTIER" not in output
def test_advisor_report_surfaces_the_sweeps_own_warnings(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, quantization="int8")
sweep = SweepSpec(batch_sizes=(1,), seq_lens=(1024,), lora_ranks=(8,))
report = advise(llama_model, training, get_gpu("a100-80"), sweep)
output = _text(render_advisor_report(report, llama_model))
assert "int8" in output
assert "unquantized base" not in output
def test_ascii_glyphs_are_chosen_by_the_console_encoding() -> None:
class _Latin(Console):
@property
def encoding(self) -> str:
return "cp1252"
assert use_ascii_glyphs(_Latin()) is True
assert use_ascii_glyphs(Console()) is False
def test_report_and_estimate_agree_on_the_golden_total(
llama_model: ModelConfig, qlora_training: TrainingConfig
) -> None:
training = replace(qlora_training, batch_size=4)
report = estimate(llama_model, training, get_gpu("4090"))
output = _text(render_report(report, llama_model, get_gpu("4090"), training))
assert "30,607" in output
assert "DOES NOT FIT" in output