Compare commits

..
Author SHA1 Message Date
hsliuustc0106 9a7ea9b003 [Bugfix] Reconcile env-var inventory with post-#6217 main drift
Merge-window drift broke the new environment-variable ratchet on main
(build 14119): #6255 added the VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT read in
the diffusion engine and #6362 removed the last package reference to
ROSVOT_SOURCE_DIR, both after #6217's CI run.

- Classify VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT as PUBLIC_OMNI and document
  it in the environment-variables reference page.
- Drop the stale ROSVOT_SOURCE_DIR model-specific row now that the
  SoulX-Singer implementation is gone.
- Update the reviewed snapshot counts and disposition tables to match.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 06:06:04 +08:00
11 changed files with 26 additions and 459 deletions

No files matched your search

+3 -2
View File
@@ -82,6 +82,7 @@ depends on the installed kernels and model path.
| --- | --- | --- | --- | --- |
| `SPEAKER_SAMPLES_DIR` | Filesystem path; default `~/.cache/vllm-omni/speakers` | Speech server; read when speaker storage initializes | Environment-only setting. The directory is created; filesystem errors propagate. | Stable |
| `SPEAKER_MAX_UPLOADED` | Integer; default `1000` | Speech server; read when speaker storage initializes | Environment-only setting. A non-integer logs a warning and uses `1000`; range is not otherwise validated. | Stable |
| `VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT` | Float seconds; default `600` | Diffusion engine async-output wait in `step_streaming`; resolved per call on the request path, not at import | Environment-only setting. A non-float or `<=0` value warns once and uses the default. | Experimental |
| `VLLM_OMNI_INPUT_WAIT_TIMEOUT_S` | Float seconds; default `600`; `<=0` disables | Full-payload input coordinator, not async-chunk transfer; read when the scheduler module imports in each worker | Environment-only setting. A non-float logs a warning and uses `600`. | Stable operational control |
| `VLLM_OMNI_ORCH_MONITOR_PATH` | Filesystem path; default `<current-working-directory>/vllm_omni_orch_monitor_<timestamp>.json` | Orchestrator monitor enabled by `--enable-orch-monitor`; read when the monitor is created | Environment-only path override. Parent directories are created; write errors are logged. | Diagnostic |
| `VLLM_OMNI_VIDEO_SYNC_TIMEOUT` | Float seconds; default `600` | Synchronous Videos API; read when the API server module imports | Environment-only setting. A non-float raises `ValueError` during import. | Experimental |
@@ -177,7 +178,7 @@ collection, examples, bug reports, or logs.
## Model-specific variables
The audit found 57 variables read by a single model or pipeline family. They are
The audit found 56 variables read by a single model or pipeline family. They are
not listed as public usage options here because doing so would turn implementation
escape hatches into an accidental compatibility contract.
@@ -190,7 +191,7 @@ Every audited model-specific name has a migration disposition in the
| Request scope | 6 | Move request-varying behavior into a declared request-option schema. |
| External | 0 | Retain only when a supported third-party library owns the contract. |
| Internalize | 11 | Keep a debug or diagnostic switch out of public documentation and configuration. |
| Deprecate/remove | 7 | Remove a compatibility escape hatch that has no continuing contract. |
| Deprecate/remove | 6 | Remove a compatibility escape hatch that has no continuing contract. |
The disposition is a migration target, not a statement that the existing
environment switch is stable. Promote or request-scope work should land in
@@ -177,24 +177,8 @@ component can be reused by the next. The DiT prefetch path and its two shared
device buffers are unchanged.
After a component is offloaded, cached-but-unallocated memory is retained only
while it stays within a budget learned from the running workload and at least
5% of device capacity remains physically free. The budget is the largest
cached value observed at any component boundary times a margin, so it tracks
the weight-reuse payload plus the transient footprint whatever the
resolution. It starts at zero: a cold start pays one release per observation
step while the peak climbs, after which boundaries retain steadily, and a
workload change whose cached footprint grows by more than the margin pays one
further release while the new peak lands. Models whose request geometry opts
out of the generic dummy run declare a feature-neutral `dummy_run_recipe`
instead; MiniMax-H3 declares a maximum-duration t2va generation, so the
engine's startup warmup runs it and, under DLO, the cold-start flushes land
before real traffic. The budget is capped at a quarter
of device capacity so a spurious observation cannot monopolize a small
device, and detaching the last attached stager resets the learned peak so one
workload never permanently influences another. No user-chosen fraction, no
multiplier constant, and no warmup flag are needed. Each decision logs the
staged byte total, observed peak, effective budget and cap,
cached-but-unallocated bytes, and free memory.
while it is at most 25% of device capacity and at least 5% of device capacity
remains physically free. Crossing either bound releases the allocator cache.
Missing allocator telemetry also releases it conservatively. Component or
staging failure forces a release, and an out-of-memory allocation gets one
retry after release. This is a component-local policy rather than a global
@@ -67,16 +67,6 @@ omni = Omni(
| `--host-weight-runtime-root PATH` | Writable node-local HWR store shared by workers in one storage domain; required for `preferred` and `required` | unset |
| `--dlo-host-registration-limit-gib N` | Optional per-worker ceiling for registering an HWR mmap; zero adds no ceiling | `0` |
## Startup warmup
Models whose request geometry needs model-specific sampling arguments for a
warmup generation (currently MiniMax-H3: `t2va`, an explicit aspect ratio)
run one at engine startup whatever features are enabled. It runs at the
maximum output duration so it covers every production request; under DLO
this means the component allocator-cache retention policy learns its
footprint during startup and no real request pays the cold-start flushes.
The tradeoff is one maximum-length generation added to boot time.
## Host-weight loading
The diffusion loader chooses host storage before DLO is enabled. It first
+3 -3
View File
@@ -164,10 +164,10 @@ def test_inventory_matches_reviewed_snapshot_counts():
"""Make an inventory expansion an explicit review decision."""
category_counts = Counter(item.category for item in ENVIRONMENT_VARIABLE_INVENTORY.values())
assert category_counts == {
EnvironmentVariableCategory.PUBLIC_OMNI: 22,
EnvironmentVariableCategory.PUBLIC_OMNI: 23,
EnvironmentVariableCategory.INHERITED_VLLM: 20,
EnvironmentVariableCategory.PLATFORM_EXTERNAL: 27,
EnvironmentVariableCategory.MODEL_SPECIFIC: 57,
EnvironmentVariableCategory.MODEL_SPECIFIC: 56,
EnvironmentVariableCategory.BENCHMARK_TRANSITIONAL: 20,
EnvironmentVariableCategory.INTERNAL: 2,
}
@@ -182,7 +182,7 @@ def test_inventory_matches_reviewed_snapshot_counts():
ModelEnvironmentVariableDisposition.REQUEST_SCOPE: 6,
ModelEnvironmentVariableDisposition.EXTERNAL: 0,
ModelEnvironmentVariableDisposition.INTERNALIZE: 11,
ModelEnvironmentVariableDisposition.DEPRECATE_REMOVE: 7,
ModelEnvironmentVariableDisposition.DEPRECATE_REMOVE: 6,
}
@@ -202,96 +202,21 @@ def test_allocator_cache_policy_enforces_cache_and_free_memory_bounds(
expected_release,
):
empty_cache, _ = patched_runtime
telemetry = {"reserved": 20, "allocated": 10, "free": 80}
monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: reserved)
monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: allocated)
monkeypatch.setattr(
residency_module.current_omni_platform,
"get_device_memory",
lambda _device: (telemetry["free"], 400),
lambda _device: (free, 100),
)
retention = BoundedAllocatorCache(torch.device("cpu"))
# Prime the learned budget: the cold-start check releases (budget 0) and
# records cached = 10, so later checks retain at budget 12.5.
assert retention.release_if_needed()
assert retention.observed_cached_peak == 10
empty_cache.reset_mock()
telemetry.update(reserved=reserved, allocated=allocated, free=free)
released = retention.release_if_needed()
assert released is expected_release
assert empty_cache.call_count == int(expected_release)
def test_attached_stagers_form_a_detach_ledger_that_resets_the_observed_peak(patched_runtime):
left, right = _aliased_modules()
retention = BoundedAllocatorCache(torch.device("cpu"))
assert retention.budget_bytes == 0
stager = PinnedModuleStager(
[left, right],
torch.device("cpu"),
pin_memory=False,
cache_retention=retention,
)
expected_bytes = 12 * 4 + 8 * 8 # float32 parameter + int64 buffer storages
assert stager.staged_bytes == expected_bytes
assert retention._staged_bytes == expected_bytes
# Re-attaching the same cache does not double count.
stager.set_cache_retention(retention)
assert retention._staged_bytes == expected_bytes
# A second stager adds its own staged bytes to the ledger.
extra = PinnedModuleStager(torch.nn.Linear(2, 2), torch.device("cpu"), pin_memory=False)
extra.set_cache_retention(retention)
assert retention._staged_bytes == expected_bytes + extra.staged_bytes
# A learned peak survives a partial detach...
retention._observed_cached_peak = 30
extra.set_cache_retention(None)
assert retention.observed_cached_peak == 30
assert retention.budget_bytes == 37
# ...but the last detach resets it so one workload cannot influence the next.
stager.set_cache_retention(None)
assert retention.observed_cached_peak == 0
assert retention.budget_bytes == 0
def test_allocator_cache_budget_is_capped_by_device_capacity(monkeypatch, patched_runtime):
empty_cache, _ = patched_runtime
telemetry = {"reserved": 40, "allocated": 10, "free": 80}
monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
monkeypatch.setattr(
residency_module.current_omni_platform,
# Capacity 100 keeps the 25% cap at 25, below the learned budget of
# 30 * 1.25 = 37.5, so the cap must govern.
"get_device_memory",
lambda _device: (telemetry["free"], 100),
)
retention = BoundedAllocatorCache(torch.device("cpu"))
# Cold start: budget 0 releases and records the peak of 30.
assert retention.release_if_needed()
assert retention.observed_cached_peak == 30
empty_cache.reset_mock()
# The learned budget 37.5 is capped to 25, so cached 30 keeps releasing.
assert retention.release_if_needed()
assert retention.release_if_needed()
empty_cache.reset_mock()
# Within the cap (cached 20 <= 25) the policy retains.
telemetry.update(reserved=30, allocated=10)
assert not retention.release_if_needed()
assert empty_cache.call_count == 0
def test_allocator_cache_releases_when_telemetry_is_unavailable(monkeypatch, patched_runtime):
empty_cache, _ = patched_runtime
@@ -303,29 +228,3 @@ def test_allocator_cache_releases_when_telemetry_is_unavailable(monkeypatch, pat
assert retention.release_if_needed()
empty_cache.assert_called_once_with()
def test_allocator_cache_budget_adapts_to_observed_peak(monkeypatch, patched_runtime):
empty_cache, _ = patched_runtime
telemetry = {"reserved": 40, "allocated": 10, "free": 80}
monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
monkeypatch.setattr(
residency_module.current_omni_platform,
"get_device_memory",
# Capacity 400 keeps the 25% cap (100) above the learned budget so
# this test isolates the peak-adaptation behavior from the cap.
lambda _device: (telemetry["free"], 400),
)
retention = BoundedAllocatorCache(torch.device("cpu"))
# Cold-start boundary: budget 0, so the policy releases and records the
# observed peak.
assert retention.release_if_needed()
assert retention.observed_cached_peak == 30
# The observation raises the budget to 30 * 1.25 = 37, so the same cached
# level is retained on every later boundary without another release.
assert not retention.release_if_needed()
assert not retention.release_if_needed()
assert empty_cache.call_count == 1
@@ -70,161 +70,6 @@ def test_dummy_run_image_count_resolves_hunyuan_architecture_alias() -> None:
assert io_support.get_dummy_run_num_image_inputs("unknown") == 1
def test_dummy_run_recipe_returns_none_without_declaration(monkeypatch: pytest.MonkeyPatch) -> None:
class PlainModel:
pass
monkeypatch.setattr(
io_support.DiffusionModelRegistry,
"_try_load_model_cls",
lambda model_class_name: PlainModel,
)
assert io_support.get_dummy_run_recipe("plain") is None
def test_dummy_run_recipe_returns_copy_of_declaration(monkeypatch: pytest.MonkeyPatch) -> None:
declared = {"task": "t2va", "duration": 4.0}
class RecipeModel:
dummy_run_recipe = declared
monkeypatch.setattr(
io_support.DiffusionModelRegistry,
"_try_load_model_cls",
lambda model_class_name: RecipeModel,
)
recipe = io_support.get_dummy_run_recipe("recipe")
assert recipe == declared
assert recipe is not declared
def test_minimax_h3_declares_a_valid_dlo_warmup_recipe() -> None:
from vllm_omni.diffusion.models.minimax_h3.pipeline_minimax_h3 import (
MINIMAX_H3_MAX_OUTPUT_SECONDS,
MiniMaxH3Pipeline,
)
recipe = MiniMaxH3Pipeline.dummy_run_recipe
assert recipe is not None
assert recipe["task"] == "t2va"
assert recipe["aspect_ratio"] == "16:9"
# Maximum duration makes the learned peak dominate every production
# request at the fixed short edge, so no real request re-learns.
assert float(recipe["duration"]) == MINIMAX_H3_MAX_OUTPUT_SECONDS
assert MiniMaxH3Pipeline.dummy_run_num_frames == 0 # generic warmup stays opted out
def _dlo_recipe_engine(enable_dlo: bool) -> DiffusionEngine:
engine = object.__new__(DiffusionEngine)
engine.od_config = SimpleNamespace(
model_class_name="RecipeModel",
enable_distributed_layerwise_offload=enable_dlo,
)
return engine
def _opt_out_generic_warmup(
monkeypatch: pytest.MonkeyPatch,
recipe: dict | None,
*,
supports_image_input: bool = False,
) -> None:
"""Make the generic dummy run opt out and stub the declared recipe."""
monkeypatch.setattr(
"vllm_omni.diffusion.diffusion_engine.supports_multimodal_input",
lambda od_config: (supports_image_input, False),
)
monkeypatch.setattr(
"vllm_omni.diffusion.diffusion_engine.get_dummy_run_num_frames",
lambda model_class_name, supports_audio_input: 0,
)
monkeypatch.setattr(
"vllm_omni.diffusion.diffusion_engine.get_dummy_run_recipe",
lambda model_class_name: recipe,
)
def test_make_dummy_request_applies_recipe_without_dlo(monkeypatch: pytest.MonkeyPatch) -> None:
# The recipe is feature-neutral: a model whose geometry needs it warms up
# whatever features are enabled, not only under DLO.
_opt_out_generic_warmup(monkeypatch, {"task": "t2va", "duration": 4.0, "aspect_ratio": "16:9"})
request = _dlo_recipe_engine(enable_dlo=False)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
assert request is not None
assert request.sampling_params.extra_args["task"] == "t2va"
def test_make_dummy_request_builds_recipe_request_when_generic_opted_out(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_opt_out_generic_warmup(
monkeypatch,
{
"task": "t2va",
"duration": 4.0,
"aspect_ratio": "16:9",
"num_inference_steps": 2,
"guidance_scale": 1.0,
},
)
request = _dlo_recipe_engine(enable_dlo=True)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
assert request is not None
assert request.is_dummy_run()
assert request.sampling_params.num_inference_steps == 2
assert request.sampling_params.guidance_scale == 1.0
extra = request.sampling_params.extra_args
assert extra["task"] == "t2va"
assert extra["duration"] == 4.0
assert extra["aspect_ratio"] == "16:9"
assert extra["cfg_text_scale"] == 1.0
def test_make_dummy_request_recipe_stays_text_only_with_image_capable_model(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# H3 advertises image input for fl2va, but the warmup recipe names t2va,
# which rejects image conditions; the recipe request must not inherit the
# generic dummy run's synthetic multimodal data.
_opt_out_generic_warmup(
monkeypatch,
{"task": "t2va", "duration": 4.0, "aspect_ratio": "16:9"},
supports_image_input=True,
)
request = _dlo_recipe_engine(enable_dlo=True)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
assert request is not None
assert not request.prompt.get("multi_modal_data")
def test_dummy_run_sends_recipe_request_when_generic_opted_out(monkeypatch: pytest.MonkeyPatch) -> None:
_opt_out_generic_warmup(monkeypatch, {"task": "t2va", "duration": 15.0, "aspect_ratio": "16:9"})
engine = _dlo_recipe_engine(enable_dlo=True)
engine.add_req_and_wait_for_response = Mock(return_value=SimpleNamespace(error=None, request_id="dlo-dummy"))
engine._dummy_run()
sent = engine.add_req_and_wait_for_response.call_args.args[0]
assert sent.is_dummy_run()
assert sent.sampling_params.extra_args["task"] == "t2va"
def test_dummy_run_still_skips_without_recipe(monkeypatch: pytest.MonkeyPatch) -> None:
_opt_out_generic_warmup(monkeypatch, None)
engine = _dlo_recipe_engine(enable_dlo=True)
engine.add_req_and_wait_for_response = Mock()
engine._dummy_run()
engine.add_req_and_wait_for_response.assert_not_called()
def test_dense_mode_does_not_build_kv_profile_request() -> None:
engine = object.__new__(DiffusionEngine)
engine.od_config = SimpleNamespace(diffusion_kv_mode=DiffusionKVCacheMode.DENSE_LEGACY)
@@ -63,6 +63,7 @@ _PUBLIC_OMNI = (
"OMNI_DIFFUSION_SESSION_STATE_MANAGER_MAX_SESSIONS",
"SPEAKER_MAX_UPLOADED",
"SPEAKER_SAMPLES_DIR",
"VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT",
"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S",
"VLLM_OMNI_ORCH_MONITOR_PATH",
"VLLM_OMNI_SERVER_STORAGE__FILE_CONCURRENCY",
@@ -197,7 +198,6 @@ _MODEL_INTERNALIZE = (
_MODEL_DEPRECATE_REMOVE = (
"HIGGS_AUDIO_V2_TOKENIZER_PATH",
"ROSVOT_SOURCE_DIR",
"VLLM_OMNI_LINGBOT_ACTION_ROOT",
"VLLM_OMNI_VOXCPM_CODE_PATH",
"VLLM_USE_FLASHINFER_MOE_FP16",
+4 -33
View File
@@ -36,7 +36,6 @@ from vllm_omni.diffusion.executor.abstract import DiffusionExecutor
from vllm_omni.diffusion.io_support import (
get_dummy_run_num_frames,
get_dummy_run_num_image_inputs,
get_dummy_run_recipe,
image_color_format,
supports_audio_output,
supports_multimodal_input,
@@ -65,9 +64,6 @@ logger = init_logger(__name__)
_ASYNC_OUTPUT_TIMEOUT_ENV = "VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT"
_ASYNC_OUTPUT_TIMEOUT_DEFAULT = 600.0 # seconds
# Warmup-recipe keys applied to the dummy request's sampling params; every
# other recipe entry is forwarded through the request's extra_args.
_DUMMY_RUN_RECIPE_SAMPLING_KEYS = frozenset({"height", "width", "num_inference_steps", "num_frames", "guidance_scale"})
def _async_output_timeout() -> float:
@@ -1021,36 +1017,8 @@ class DiffusionEngine:
) -> OmniDiffusionRequest | None:
"""Build a one-step model request for startup profiling or warmup."""
supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config)
num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input)
if num_frames <= 0:
# The model opted out of the generic dummy run; fall back to its
# declared warmup recipe, which supplies whatever feature-specific
# sampling arguments the model requires (under distributed
# layerwise offload the generation also primes the component
# allocator-cache retention policy ahead of real traffic). The
# recipe names its own task, so the request stays text-only even
# when the model's other tasks accept image or audio conditions.
recipe = get_dummy_run_recipe(self.od_config.model_class_name)
if recipe is None:
return None
extra_args: dict[str, Any] = {"cfg_text_scale": 1.0, "cfg_img_scale": 1.0}
extra_args.update({k: v for k, v in recipe.items() if k not in _DUMMY_RUN_RECIPE_SAMPLING_KEYS})
return OmniDiffusionRequest(
prompt={"prompt": "dummy run"},
request_id=DUMMY_DIFFUSION_REQUEST_ID,
sampling_params=OmniDiffusionSamplingParams(
height=int(recipe.get("height", height)),
width=int(recipe.get("width", width)),
num_inference_steps=int(recipe.get("num_inference_steps", 1)),
num_frames=int(recipe.get("num_frames", 0)),
guidance_scale=float(recipe.get("guidance_scale", guidance_scale)),
num_outputs_per_prompt=1,
extra_args=extra_args,
),
)
prompt: OmniTextPrompt = {"prompt": "dummy run"}
supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config)
if supports_image_input:
color_format = image_color_format(self.od_config.model_class_name)
images = [PIL.Image.new(color_format, (width, height)) for _ in range(num_image_inputs)]
@@ -1060,6 +1028,9 @@ class DiffusionEngine:
audio_sr = 16000
prompt.setdefault("multi_modal_data", {})["audio"] = np.random.randn(audio_sr * 2).astype(np.float32)
num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input)
if num_frames <= 0:
return None
return OmniDiffusionRequest(
prompt=prompt,
request_id=DUMMY_DIFFUSION_REQUEST_ID,
-17
View File
@@ -4,7 +4,6 @@
from __future__ import annotations
import inspect
from typing import Any
from vllm_omni.diffusion.data import OmniDiffusionConfig
from vllm_omni.diffusion.model_metadata import get_diffusion_model_metadata
@@ -55,22 +54,6 @@ def get_dummy_run_num_frames(model_class_name: str, supports_audio_input: bool)
return 2 if supports_audio_input or supports_audio_output(model_class_name) else 1
def get_dummy_run_recipe(model_class_name: str) -> dict[str, Any] | None:
"""Get the model-declared startup-warmup recipe, or None.
Models that opt out of the generic dummy run (``dummy_run_num_frames = 0``)
because their request geometry needs model-specific sampling arguments
(task, duration, aspect ratio, ...) declare a recipe instead. The recipe
is feature-neutral: the engine runs it at startup whatever features are
enabled — under distributed layerwise offload the generation also primes
the component allocator-cache retention policy's observed-peak budget.
"""
model_cls = DiffusionModelRegistry._try_load_model_cls(model_class_name)
recipe = getattr(model_cls, "dummy_run_recipe", None) if model_cls is not None else None
return dict(recipe) if recipe else None
def get_dummy_run_num_image_inputs(model_class_name: str) -> int:
"""Return the maximum advertised image-input count for profiling."""
@@ -704,20 +704,6 @@ class MiniMaxH3Pipeline(
"post_decode",
]
dummy_run_num_frames: ClassVar[int] = 0
# Startup warmup recipe: the generic dummy run cannot satisfy t2va's
# required sampling arguments (explicit aspect ratio, bounded duration).
# Duration is the only workload variable at the fixed 768 short edge, so
# warming at the maximum covers every production request — under DLO this
# makes the learned cache-retention peak dominate, moving all cold-start
# flushes ahead of real traffic. The cost is one max-length generation in
# boot time whatever features are enabled.
dummy_run_recipe: ClassVar[Mapping[str, Any] | None] = {
"task": "t2va",
"duration": MINIMAX_H3_MAX_OUTPUT_SECONDS,
"aspect_ratio": "16:9",
"num_inference_steps": 2,
"guidance_scale": 1.0,
}
# Only distilled releases pin a schedule, so the default keeps the legacy
# uniform path available to partially constructed pipelines.
_base_schedule_by_partition: ClassVar[Mapping[str, DMD2SigmaSchedule | None]] = {}
+10 -102
View File
@@ -20,17 +20,6 @@ from .tensor_utils import is_dtensor, set_tensor_storage
logger = init_logger(__name__)
# Margin over the largest cached-but-unallocated value observed at a component
# boundary. The observed peak already includes the weight-reuse payload and
# the transient footprint of the running workload, so this only covers
# check-to-check variation.
_OBSERVED_PEAK_MARGIN = 1.25
# Hard ceiling for the learned budget so a spurious observation can never
# monopolize a small device; the #6526 L20X retention plateau (~23.8 GB on a
# ~140 GB card) stayed well below this fraction.
_MAX_BUDGET_CAPACITY_FRACTION = 0.25
class BoundedAllocatorCache:
"""Retain reusable allocator blocks without monopolizing device memory.
@@ -39,100 +28,35 @@ class BoundedAllocatorCache:
cached blocks are immediately reusable. This policy keeps the cache while
both of these bounds hold:
* cached-but-unallocated memory stays within the learned budget — the
largest cached value observed at any component boundary times a margin,
capped at a quarter of device capacity; and
* cached-but-unallocated memory is at most 25% of device capacity; and
* at least 5% of device capacity is physically free.
The budget needs no user-chosen fraction, no multiplier constant, and no
warmup flag. It starts at zero, so the first boundaries of a cold start
release once per observation step while the peak climbs; afterwards the
budget tracks the observed peak and boundaries retain steadily. A
workload change whose cached footprint grows by more than the margin pays
one further release while the new peak lands. Detaching the last attached
``PinnedModuleStager`` resets the observed peak, so one workload never
permanently influences another. Missing memory telemetry is handled
conservatively by releasing the cache. Failure paths can force release;
normal executor shutdown keeps its own unconditional device-cache cleanup
because this policy is not global.
Missing memory telemetry is handled conservatively by releasing the cache.
Failure paths can force release; normal executor shutdown keeps its own
unconditional device-cache cleanup because this policy is not global.
"""
def __init__(
self,
device: torch.device,
*,
max_cached_fraction: float = 0.25,
min_free_fraction: float = 0.05,
) -> None:
if not 0.0 <= max_cached_fraction <= 1.0:
raise ValueError(f"max_cached_fraction must be in [0, 1], got {max_cached_fraction}")
if not 0.0 <= min_free_fraction <= 1.0:
raise ValueError(f"min_free_fraction must be in [0, 1], got {min_free_fraction}")
self.device = device
self.max_cached_fraction = max_cached_fraction
self.min_free_fraction = min_free_fraction
self._staged_bytes = 0
self._observed_cached_peak = 0
def grow_retention_budget(self, staged_bytes: int) -> None:
"""Record an attached stager's staged byte total."""
if staged_bytes < 0:
raise ValueError(f"staged_bytes must be >= 0, got {staged_bytes}")
self._staged_bytes += int(staged_bytes)
def shrink_retention_budget(self, staged_bytes: int) -> None:
"""Record a stager detach; the last detach resets the observed peak."""
if staged_bytes < 0:
raise ValueError(f"staged_bytes must be >= 0, got {staged_bytes}")
self._staged_bytes = max(0, self._staged_bytes - int(staged_bytes))
if self._staged_bytes == 0:
self._observed_cached_peak = 0
@property
def observed_cached_peak(self) -> int:
"""Largest cached-but-unallocated value seen at a boundary check."""
return self._observed_cached_peak
@property
def budget_bytes(self) -> int:
"""Reusable-cache ceiling: the observed peak times its margin."""
return int(self._observed_cached_peak * _OBSERVED_PEAK_MARGIN)
def _should_release(self) -> bool:
reserved = int(torch.accelerator.memory_reserved(self.device))
allocated = int(torch.accelerator.memory_allocated(self.device))
free, total = current_omni_platform.get_device_memory(self.device)
cached = max(0, reserved - allocated)
capacity_cap = int(total * _MAX_BUDGET_CAPACITY_FRACTION)
budget = min(self.budget_bytes, capacity_cap)
# Decide against the pre-observation budget, then record the peak so a
# one-off over-budget release also teaches the budget for later checks.
release = cached > budget or free < int(total * self.min_free_fraction)
if cached > self._observed_cached_peak:
self._observed_cached_peak = cached
if release:
reason = "cached-over-budget" if cached > budget else "low-free-memory"
logger.debug(
"Releasing retained allocator cache (%s): cached=%d budget=%d staged=%d "
"observed_peak=%d capacity_cap=%d free=%d total=%d",
reason,
cached,
budget,
self._staged_bytes,
self._observed_cached_peak,
capacity_cap,
free,
total,
)
return True
logger.debug(
"Retaining allocator cache: cached=%d budget=%d staged=%d observed_peak=%d "
"capacity_cap=%d free=%d total=%d",
cached,
budget,
self._staged_bytes,
self._observed_cached_peak,
capacity_cap,
free,
total,
)
return False
return cached > int(total * self.max_cached_fraction) or free < int(total * self.min_free_fraction)
def release_if_needed(self, *, force: bool = False) -> bool:
"""Release cached blocks when a bound is crossed or release is forced."""
@@ -192,12 +116,11 @@ class PinnedModuleStager:
self.device = device
self.copy_stream = copy_stream if copy_stream is not None else current_omni_platform.Stream()
self._ready_event = current_omni_platform.Event()
self.cache_retention = None
self.cache_retention = cache_retention
self.loaded = False
self._groups = self._snapshot_groups(modules, pin_memory=pin_memory)
self._device_storages: list[torch.Tensor] = []
self._restore_masters()
self.set_cache_retention(cache_retention)
@staticmethod
def _local_tensor(target: torch.Tensor) -> torch.Tensor:
@@ -275,19 +198,8 @@ class PinnedModuleStager:
def _restore_masters(self) -> None:
self._bind([group.master for group in self._groups])
@property
def staged_bytes(self) -> int:
"""Total unique staged storage bytes (masters are uint8 byte views)."""
return sum(group.master.numel() for group in self._groups)
def set_cache_retention(self, cache_retention: BoundedAllocatorCache | None) -> None:
if cache_retention is self.cache_retention:
return
if self.cache_retention is not None:
self.cache_retention.shrink_retention_budget(self.staged_bytes)
self.cache_retention = cache_retention
if cache_retention is not None:
cache_retention.grow_retention_budget(self.staged_bytes)
def _release_cache(self, *, force: bool = False) -> None:
if self.cache_retention is None:
@@ -327,10 +239,6 @@ class PinnedModuleStager:
except torch.OutOfMemoryError:
# A retained cache is normally reusable by this process, but an
# explicit flush gives external memory pressure one bounded retry.
logger.info(
"Staging out of memory (%d staged bytes); releasing retained cache and retrying once",
self.staged_bytes,
)
self._cleanup_failed_load()
try:
self._load_once()