Compare commits

..
Author SHA1 Message Date
hsliuustc0106 e1c4b6153a [Model] Make the dummy-run recipe feature-neutral
Per review, the warmup recipe is a property of the model's request
geometry, not of distributed layerwise offload: rename
dlo_dummy_run_recipe to dummy_run_recipe and drop the DLO gate in
_make_dummy_request. A model whose geometry needs recipe-specific sampling
arguments warms up at startup whatever features are enabled — under DLO
the generation also primes the component cache-retention budget; without
it, the same warmup covers any other first-request cost. Validated both
ways on L20X: DLO boot unchanged (releases stay in the startup window) and
a non-DLO MiniMax-H3 boot now runs the warmup and serves normally.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 15:48:07 +08:00
hsliuustc0106 33297bdb9d [Model] Fold the DLO warmup recipe into _make_dummy_request
One request builder instead of two: when the generic dummy run opts out
(num_frames <= 0), _make_dummy_request itself falls back to the model's
declared dlo_dummy_run_recipe under distributed layerwise offload, and
_dummy_run keeps its original shape. The recipe request is text-only even
for image-capable models — the recipe names its own task, and t2va rejects
image conditions; caught by a boot smoke after folding (the standalone
builder had built the prompt fresh) and pinned by a regression test.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 15:35:13 +08:00
hsliuustc0106 f4394f1d3d [Model] Prime DLO cache retention with a startup warmup
Models whose request geometry opts out of the generic dummy run (H3: t2va
needs an explicit aspect ratio and a bounded duration) can declare a
dlo_dummy_run_recipe. Under distributed layerwise offload the engine's
startup warmup runs it, so the learned cache-retention budget is primed
before real traffic instead of on the first request.

H3 declares a maximum-duration t2va generation: duration is the only
workload variable at the fixed 768 short edge, so the learned peak
dominates every production request and no real request re-learns. Validated
on L20X: all cold-start releases happen during the startup warmup, zero
during requests; first request 15.91 s vs 16.7 s unwarmed, steady 14.9 s;
startup gains one max-length generation (~10-20 s). Non-DLO deployments
keep the opted-out generic warmup unchanged.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 15:06:08 +08:00
hsliuustc0106 c9bb65902e [Model] Learn the retention budget directly from observed peaks
Drop the staged-bytes seed per review direction: the budget is now
min(observed_peak * 1.25, 25% of device capacity), nothing else. A cold
start pays one release per observation step while the peak climbs (two on
the validated MiniMax workload), then boundaries retain steadily; the 5%
free floor and capacity cap are unchanged. The staged-byte ledger remains
only to detect full detachment, which now resets the observed peak so one
workload cannot permanently influence another. Trade-off accepted per
review: lower reserved memory and a simpler policy over a flush-free
first request.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 13:43:00 +08:00
hsliuustc0106 b78db6036a [Model] Fix adaptation test fixture so the capacity cap does not bind
The fixture reported device capacity 100, making the 25% cap 25 — below
the learned budget of 37 — so the post-adaptation checks released again
and the test failed. Raise the fixture capacity to 400 so the cap stays
above the learned budget and the test isolates peak adaptation; verified
against the real suite on the L20X validation host (15 passed).

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 13:12:55 +08:00
hsliuustc0106 27f9726a6a [Model] Derive the retention budget from observed boundary peaks
The staged-bytes seed left the transient half of the working set to a
guessed multiplier, and that half scales with the running workload rather
than the model topology. Track the largest cached-but-unallocated value
observed at any component boundary instead: the budget is the larger of
the seed and the observed peak times a 1.25 margin, still capped at a
quarter of device capacity. A workload that outgrows the seed pays one
release while the observation lands, then retains steadily. No warmup
flag is needed and no user-chosen fraction exists.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 12:52:19 +08:00
hsliuustc0106 c5295f3ffe [Model] Cap derived cache budget and account for stager detach
Address review on #6629:

- Cap the derived retention budget at a quarter of device capacity so a
  large staged total cannot monopolize a small device; the multiplier
  itself stays pending an instrumented target run.
- Log every release decision (reason, cached bytes, effective budget and
  cap, staged total, free/total memory) and every staging OOM retry with
  its staged byte total, so a single target run validates the headroom
  multiplier directly.
- Make stager detach symmetric: set_cache_retention now shrinks the
  previous cache's budget before growing the new one, so cache -> None ->
  same cache counts the staged bytes exactly once (regression test added).

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 11:12:09 +08:00
hsliuustc0106 ba5943dddb [Model] Derive MiniMax-H3 DLO component cache budget from staged bytes
Replace BoundedAllocatorCache's device-capacity fraction bound (25% of
total memory) with a budget derived from the model topology: each attached
PinnedModuleStager reports its unique staged storage bytes once, and the
retention ceiling is that total times a fixed 2.0 headroom multiplier
covering allocator fragmentation and transient storages freed at component
boundaries. The 5% physically-free floor is unchanged.

The budget therefore scales with the staged components instead of device
capacity, so no user-chosen fraction is needed and behavior is identical
across differently sized devices with the same model. Re-attaching the
same cache does not double count; a second stager grows the budget by its
own bytes.

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>
2026-08-26 09:49:55 +08:00
20 changed files with 474 additions and 343 deletions

No files matched your search

@@ -29,7 +29,6 @@ Legend: ✅ supported, ⚠️ compatibility path or limited validation, ❌ unsu
| **Other online quantization methods** | ❌ Rejected until runtime packing and scale layouts are validated. | ⚠️ Allowed through the ordinary loader; validation is method-specific. |
| **Model-level or standard layerwise CPU offload** | ❌ Disabled because DLO takes priority. | ❌ Disabled because DLO takes priority. |
| **Resident leading layers** | ❌ Rejected. | ✅ Requires eligible resident paths in the model's `OffloadPlan`. |
| **Auxiliary component policy** | ✅ Encoders/VAEs may remain resident. | ✅ Encoders/VAEs may remain resident. |
See [Parallelism compatibility](#parallelism-compatibility) and
[Request and loading constraints](#request-and-loading-constraints) for the
@@ -178,8 +177,24 @@ 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 is at most 25% of device capacity and at least 5% of device capacity
remains physically free. Crossing either bound releases the allocator cache.
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.
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
@@ -420,22 +435,6 @@ failure.
- **HSDP + DP or TP:** rejected independently by the diffusion parallel
configuration.
### Auxiliary component policy
`dlo_offload_components` maps discovered encoder/VAE paths to booleans. A
component set to `False` is materialized on the target device once and is not
entered through the model's manual offload lifecycle. A `True` value permits
the existing `OffloadPlan.on_demand_component_paths` and encoder block-hook
behavior; it does not add support where the model declares none. The optional
`default` key handles unmatched auxiliary components, and an empty map
preserves the current policy.
The policy is resolved after component discovery and before DLO mutates DiT
storage. Unknown names, including DiT paths, are rejected. DLO continues to
own DiT streaming regardless of this map, which avoids an invalid state where
the loader has skipped ordinary DiT materialization but the backend is asked
to keep that DiT resident.
## Request and loading constraints
AllGather DP multi-concurrency requires:
@@ -447,15 +446,6 @@ AllGather DP multi-concurrency requires:
The no-AllGather path does not impose these DLO-specific synchronized-wave
requirements.
MiniMax-H3 currently builds one global encoder TP group and broadcasts the
resulting conditioning from global rank 0. Therefore an encoder TP group that
spans multiple DP replicas is not valid for independent-prompt AllGather
waves. Exact-recipe DP2/TP2 and DP2/SP2 experiments with encoder TP4 completed
the collective schedule, but two distinct prompts with the same seed produced
byte-identical video and audio in every measured wave. Enabling or disabling
encoder offload produced paired-identical results, so this is a pre-existing
model topology limitation rather than component-policy behavior.
Direct checkpoint mmap can back either transfer path. It is currently limited
to proven TP1, non-HSDP, non-online-quantized layouts. Other layouts use the
ordinary loader. Per-tensor online FP8 linears can use DLO AllGather after the
@@ -520,9 +510,6 @@ AllGather, because DP replicas did not share checkpoint-backed runtime
weights. This is a functional and memory smoke test, not a production-quality
performance or output-quality benchmark.
These B300 DP waves intentionally reused the same prompt and seed, so they do
not establish independent-prompt isolation across DP replicas.
### Host-memory measurement
A two-worker MiniMax-H3 FL2VA measurement on one L20X node compared the
@@ -36,8 +36,7 @@ vllm serve /path/to/model --omni \
vllm serve /path/to/model --omni \
--enable-distributed-layerwise-offload \
--data-parallel-size 4 \
--dlo-no-use-allgather \
--dlo-offload-components '{"text_encoder": false}'
--dlo-no-use-allgather
# Sequence parallel deployment
vllm serve /path/to/model --omni \
@@ -52,7 +51,6 @@ omni = Omni(
model="/path/to/model",
enable_distributed_layerwise_offload=True,
dlo_use_allgather=True,
dlo_offload_components={"text_encoder": False},
)
```
@@ -65,11 +63,20 @@ omni = Omni(
| `--dlo-use-allgather` | Shard host weights and reconstruct with AllGather | `true` |
| `--dlo-no-use-allgather` | Stream complete rank-local blocks without a DLO weight collective | `false` |
| `--dlo-resident-layers N` | Keep N leading main-DiT blocks on device; requires no-AllGather and model-declared resident paths | `0` |
| `--dlo-offload-components JSON` | Permit or disable offload for discovered encoder/VAE components | `{}` |
| `--host-weight-runtime-mode {disabled,preferred,required}` | HWR policy: no interaction, populate on a miss, or require an exact hit | `disabled` |
| `--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
@@ -247,25 +254,6 @@ class MyPipeline(nn.Module):
When no plan exists, discovery falls back to
`_layerwise_offload_blocks_attrs` and then heuristic attribute lookup.
## Auxiliary component policy
`dlo_offload_components` follows the same component-map style as diffusion
quantization. Keys are discovered encoder or VAE paths, values are booleans,
and `default` is the optional fallback:
```python
dlo_offload_components = {
"text_encoder": False,
"default": True,
}
```
`True` permits the model's existing DLO behavior for that component. `False`
keeps the whole component resident, including its child blocks and non-block
modules. An empty map preserves current behavior. Unknown keys are rejected;
the map does not control DiT blocks because enabling DLO gives the backend
ownership of DiT streaming.
## Data-parallel concurrency
With `data_parallel_size > 1` and AllGather enabled, the scheduler can process
@@ -290,13 +278,6 @@ must enter each collective.
are validated.
- Resident leading layers require `--dlo-no-use-allgather` and a model
`OffloadPlan` that declares eligible `resident_dit_paths`.
- Component policy applies only to discovered auxiliary encoders and VAEs;
it cannot disable DLO for a DiT.
- MiniMax-H3's encoder TP group is global rather than DP-replica-local. An
encoder TP size that spans multiple DP replicas (for example TP4 with
DP2/TP2 or DP2/SP2 on four ranks) is not supported for independent-prompt
AllGather waves: the current pipeline broadcasts rank 0's conditioning to
both replicas. This limitation is independent of the component policy.
- DP concurrency requires an explicit, identical inference-step count.
Sharing quantized or otherwise unvalidated transformed layouts through a
@@ -1,5 +1,5 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Validate MiniMax-H3 request-mode and DLO lifecycle behavior.
The DLO modes accept any positive data-parallel size. With AllGather enabled,
@@ -85,20 +85,12 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--height", type=int, default=768)
parser.add_argument("--batch-wait-ms", type=float, default=500.0)
parser.add_argument("--init-timeout", type=float, default=1800.0)
parser.add_argument(
"--dlo-offload-components",
type=json.loads,
default=None,
help="Component policy JSON, e.g. '{\"text_encoder\": false}'",
)
parser.add_argument("--output", type=Path)
return parser.parse_args()
def engine_kwargs(args: argparse.Namespace) -> dict[str, Any]:
is_dlo = args.mode in {"dlo", "dlo-no-allgather"}
if args.dlo_offload_components and not is_dlo:
raise ValueError("--dlo-offload-components requires a DLO mode")
common: dict[str, Any] = {
"model": args.model,
"trust_remote_code": True,
@@ -122,7 +114,6 @@ def engine_kwargs(args: argparse.Namespace) -> dict[str, Any]:
enable_distributed_layerwise_offload=True,
dlo_use_allgather=args.mode == "dlo",
dlo_resident_layers=0,
dlo_offload_components=({} if args.dlo_offload_components is None else args.dlo_offload_components),
)
else:
common.update(
@@ -99,22 +99,3 @@ def test_manual_component_offload_failure_forces_retained_cache_release(mocker):
assert component.load_to_device.call_count == 2
assert component.offload_to_cpu.call_count == 2
pipeline._dlo_component_cache.release_if_needed.assert_called_once_with(force=True)
def test_resident_component_skips_manual_dlo_lifecycle(mocker):
from vllm_omni.diffusion.models.minimax_h3 import MiniMaxH3Pipeline
pipeline = object.__new__(MiniMaxH3Pipeline)
torch.nn.Module.__init__(pipeline)
pipeline.od_config = mocker.Mock()
pipeline.od_config.enable_layerwise_offload = False
pipeline.od_config.enable_distributed_layerwise_offload = True
pipeline._model_cpu_offload_modules = []
component = mocker.Mock()
component._omni_dlo_offload_enabled = False
with pipeline._component_on_device(component):
pass
component.load_to_device.assert_not_called()
component.offload_to_cpu.assert_not_called()
@@ -1,142 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from types import SimpleNamespace
import pytest
import torch
from torch import nn
from vllm_omni.diffusion.offloader.base import OffloadConfig, OffloadStrategy
from vllm_omni.diffusion.offloader.distributed_layerwise_backend import (
DistributedLayerwiseOffloadBackend,
)
from vllm_omni.diffusion.offloader.module_collector import PipelineModules
from vllm_omni.diffusion.offloader.offload_plan import OffloadPlan
pytestmark = [pytest.mark.diffusion, pytest.mark.cpu, pytest.mark.core_model]
def _od_config(**overrides):
values = {
"enable_cpu_offload": False,
"enable_layerwise_offload": False,
"enable_distributed_layerwise_offload": True,
"dlo_use_allgather": False,
"dlo_resident_layers": 0,
"dlo_offload_components": {},
"pin_cpu_memory": False,
"parallel_config": SimpleNamespace(
use_hsdp=False,
data_parallel_size=1,
sequence_parallel_size=1,
),
}
values.update(overrides)
return SimpleNamespace(**values)
def _backend(policy: dict[str, bool]) -> DistributedLayerwiseOffloadBackend:
backend = object.__new__(DistributedLayerwiseOffloadBackend)
backend.config = OffloadConfig(
strategy=OffloadStrategy.DISTRIBUTED_LAYER_WISE,
pin_cpu_memory=False,
dlo_use_allgather=False,
dlo_offload_components=policy,
)
backend.device = torch.device("cpu")
return backend
def _modules() -> PipelineModules:
return PipelineModules(
dits=[],
dit_names=[],
encoders=[nn.Module()],
encoder_names=["text_encoder"],
vaes=[nn.Module(), nn.Module()],
vae_names=["video_vae", "audio_vae"],
)
def test_component_policy_resolves_exact_then_default():
backend = _backend({"text_encoder": False, "default": True})
assert not backend._component_offload_enabled("text_encoder")
assert backend._component_offload_enabled("video_vae")
@pytest.mark.parametrize(
("value", "error", "match"),
[
([], TypeError, "must be a mapping"),
(["text_encoder"], TypeError, "must be a mapping"),
(False, TypeError, "must be a mapping"),
({1: False}, TypeError, "keys must be non-empty strings"),
({"text_encoder": 0}, TypeError, "must be a boolean"),
],
)
def test_component_policy_rejects_invalid_shapes(value, error, match):
with pytest.raises(error, match=match):
OffloadConfig.from_od_config(_od_config(dlo_offload_components=value))
def test_component_policy_accepts_none_as_default():
config = OffloadConfig.from_od_config(_od_config(dlo_offload_components=None))
assert config.dlo_offload_components == {}
def test_component_policy_requires_dlo():
with pytest.raises(ValueError, match="requires distributed layerwise offload"):
OffloadConfig.from_od_config(
_od_config(
enable_distributed_layerwise_offload=False,
dlo_offload_components={"text_encoder": False},
)
)
def test_component_policy_rejects_unknown_or_dit_names():
backend = _backend({"transformer": False})
with pytest.raises(ValueError, match="unknown auxiliary components.*transformer"):
backend._validate_component_offload_policy(_modules())
@pytest.mark.parametrize(
("policy", "encoder_stage", "video_stage", "audio_stage", "try_encoder"),
[
({}, True, True, True, True),
({"text_encoder": False}, False, True, True, False),
({"default": False, "video_vae": True}, False, True, False, False),
],
)
def test_prepare_auxiliary_components_applies_policy(
mocker,
policy,
encoder_stage,
video_stage,
audio_stage,
try_encoder,
):
backend = _backend(policy)
backend._try_layerwise_offload_encoder = mocker.Mock()
backend._register_on_demand_hook = mocker.Mock()
modules = _modules()
plan = OffloadPlan(
encoder_block_attrs={"text_encoder": ("layers",)},
on_demand_component_paths=frozenset({"text_encoder", "video_vae", "audio_vae"}),
)
backend._prepare_auxiliary_components(modules, plan)
assert backend._try_layerwise_offload_encoder.called is try_encoder
assert modules.encoders[0]._omni_dlo_offload_enabled is encoder_stage
assert modules.vaes[0]._omni_dlo_offload_enabled is video_stage
assert modules.vaes[1]._omni_dlo_offload_enabled is audio_stage
assert backend._register_on_demand_hook.call_args_list == [
mocker.call(modules.encoders[0], "text_encoder", stage_on_demand=encoder_stage),
mocker.call(modules.vaes[0], "video_vae", stage_on_demand=video_stage),
mocker.call(modules.vaes[1], "audio_vae", stage_on_demand=audio_stage),
]
@@ -202,21 +202,96 @@ def test_allocator_cache_policy_enforces_cache_and_free_memory_bounds(
expected_release,
):
empty_cache, _ = patched_runtime
monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: reserved)
monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: allocated)
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(
residency_module.current_omni_platform,
"get_device_memory",
lambda _device: (free, 100),
lambda _device: (telemetry["free"], 400),
)
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
@@ -228,3 +303,29 @@ 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,6 +70,161 @@ 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)
@@ -332,7 +332,7 @@ def test_serve_cli_forwards_distilled_lora_to_diffusion_stage():
]
def test_serve_cli_forwards_distributed_offload_options():
def test_serve_cli_forwards_distributed_offload_residency():
"""Ensure the two-GPU DLO placement controls reach the diffusion stage."""
parser = TrackingArgumentParser()
subparsers = parser.add_subparsers(dest="command")
@@ -347,8 +347,6 @@ def test_serve_cli_forwards_distributed_offload_options():
"--dlo-no-use-allgather",
"--dlo-resident-layers",
"20",
"--dlo-offload-components",
'{"text_encoder": false, "default": true}',
]
)
@@ -359,14 +357,9 @@ def test_serve_cli_forwards_distributed_offload_options():
assert args.enable_distributed_layerwise_offload is True
assert args.dlo_use_allgather is False
assert args.dlo_resident_layers == 20
assert args.dlo_offload_components == {"text_encoder": False, "default": True}
assert engine_args["enable_distributed_layerwise_offload"] is True
assert engine_args["dlo_use_allgather"] is False
assert engine_args["dlo_resident_layers"] == 20
assert engine_args["dlo_offload_components"] == {
"text_encoder": False,
"default": True,
}
def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():
-1
View File
@@ -697,7 +697,6 @@ class _DiffusionConfigProjection:
enable_distributed_layerwise_offload: bool = False
dlo_use_allgather: bool = True
dlo_resident_layers: int = Field(default=0, ge=0)
dlo_offload_components: dict[str, bool] = Field(default_factory=dict)
host_weight_runtime_mode: Literal["disabled", "preferred", "required"] = "disabled"
host_weight_runtime_root: str | None = None
dlo_host_registration_limit_gib: float = Field(default=0.0, ge=0)
-1
View File
@@ -389,7 +389,6 @@ class StageDeployConfig:
enable_distributed_layerwise_offload: bool | None = None
dlo_use_allgather: bool | None = None
dlo_resident_layers: int | None = None
dlo_offload_components: dict[str, bool] | None = None
host_weight_runtime_mode: str | None = None
host_weight_runtime_root: str | None = None
dlo_host_registration_limit_gib: float | None = None
-3
View File
@@ -795,9 +795,6 @@ class OmniDiffusionConfig:
dlo_use_allgather: bool = True
# Leading main-DiT blocks kept resident by distributed layerwise offload.
dlo_resident_layers: int = 0
# Auxiliary encoder/VAE policy. False keeps the whole component resident;
# DiT offload remains owned by DLO.
dlo_offload_components: dict[str, bool] = field(default_factory=dict)
# Final-layout Host Weight Runtime policy. The loader only activates this
# for eligible no-AllGather DLO; all other configurations preserve their
# existing loader/storage path.
+33 -4
View File
@@ -36,6 +36,7 @@ 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,
@@ -64,6 +65,9 @@ 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:
@@ -1017,8 +1021,36 @@ class DiffusionEngine:
) -> OmniDiffusionRequest | None:
"""Build a one-step model request for startup profiling or warmup."""
prompt: OmniTextPrompt = {"prompt": "dummy run"}
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"}
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)]
@@ -1028,9 +1060,6 @@ 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,6 +4,7 @@
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
@@ -54,6 +55,22 @@ 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,6 +704,20 @@ 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]] = {}
@@ -1436,7 +1450,7 @@ class MiniMaxH3Pipeline(
with sequential_offload_component(component):
yield
return
staged = self._uses_manual_component_offload() and getattr(component, "_omni_dlo_offload_enabled", True)
staged = self._uses_manual_component_offload()
try:
if staged:
component.load_to_device()
+1 -25
View File
@@ -2,8 +2,7 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from abc import ABC, abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass, field
from dataclasses import dataclass
from enum import Enum
from typing import Protocol, runtime_checkable
@@ -53,9 +52,6 @@ class OffloadConfig:
# blocks from the loader-selected host backing with H2D only.
dlo_use_allgather: bool = True
dlo_resident_layers: int = 0 # leading DiT layers kept on device
# Per auxiliary component: True permits model-declared DLO offload;
# False keeps the whole component resident. "default" is the fallback.
dlo_offload_components: dict[str, bool] = field(default_factory=dict)
# Optional per-worker ceiling for registering an HWR mmap. Zero means no
# additional ceiling; pin_cpu_memory controls whether registration is tried.
dlo_host_registration_limit_gib: float = 0.0
@@ -129,25 +125,6 @@ class OffloadConfig:
# requirements (concurrent requests, dummy run skip).
dlo_use_allgather = getattr(od_config, "dlo_use_allgather", True)
dlo_resident_layers = int(getattr(od_config, "dlo_resident_layers", 0))
raw_component_policy = getattr(od_config, "dlo_offload_components", {})
if raw_component_policy is None:
raw_component_policy = {}
if not isinstance(raw_component_policy, Mapping):
raise TypeError(
"dlo_offload_components must be a mapping of component names to booleans, "
f"got {type(raw_component_policy).__name__}"
)
dlo_offload_components: dict[str, bool] = {}
for component, enabled in raw_component_policy.items():
if not isinstance(component, str) or not component:
raise TypeError("dlo_offload_components keys must be non-empty strings")
if not isinstance(enabled, bool):
raise TypeError(
f"dlo_offload_components[{component!r}] must be a boolean, got {type(enabled).__name__}"
)
dlo_offload_components[component] = enabled
if dlo_offload_components and not enable_distributed_layerwise_offload:
raise ValueError("dlo_offload_components requires distributed layerwise offload")
dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
limit_gib=getattr(od_config, "dlo_host_registration_limit_gib", 0.0),
enable_dlo=enable_distributed_layerwise_offload,
@@ -189,7 +166,6 @@ class OffloadConfig:
dp_size=dp_size,
dlo_use_allgather=dlo_use_allgather,
dlo_resident_layers=dlo_resident_layers,
dlo_offload_components=dlo_offload_components,
dlo_host_registration_limit_gib=dlo_host_registration_limit_gib,
)
@@ -43,7 +43,7 @@ from .host_registration import (
HostRegistrationError,
register_host_mappings,
)
from .module_collector import ModuleDiscovery, PipelineModules
from .module_collector import ModuleDiscovery
from .offload_plan import (
OffloadPlan,
get_offload_plan,
@@ -1400,39 +1400,6 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
module.to(self.device)
logger.info("Moved %s (%s) to GPU (resident)", label, module.__class__.__name__)
def _component_offload_enabled(self, name: str) -> bool:
policy = self.config.dlo_offload_components
return policy.get(name, policy.get("default", True))
def _validate_component_offload_policy(self, modules: PipelineModules) -> None:
known_components = {"default", *modules.encoder_names, *modules.vae_names}
unknown = sorted(self.config.dlo_offload_components.keys() - known_components)
if unknown:
raise ValueError(
f"dlo_offload_components contains unknown auxiliary components: {unknown}. DLO always owns DiT offload."
)
def _prepare_auxiliary_components(self, modules: PipelineModules, plan: OffloadPlan | None) -> None:
for encoder, name in zip(modules.encoders, modules.encoder_names, strict=True):
offload = self._component_offload_enabled(name)
encoder._omni_dlo_offload_enabled = offload
if offload:
self._try_layerwise_offload_encoder(encoder, name, plan)
self._register_on_demand_hook(
encoder,
name,
stage_on_demand=(offload and plan is not None and name in plan.on_demand_component_paths),
)
for vae, name in zip(modules.vaes, modules.vae_names, strict=True):
offload = self._component_offload_enabled(name)
vae._omni_dlo_offload_enabled = offload
self._register_on_demand_hook(
vae,
name,
stage_on_demand=(offload and plan is not None and name in plan.on_demand_component_paths),
)
def _try_layerwise_offload_encoder(self, module: nn.Module, name: str, plan: OffloadPlan | None) -> bool:
"""Stream plan-declared encoder blocks on each rank without AllGather."""
if plan is None or name not in plan.encoder_block_attrs:
@@ -1701,7 +1668,6 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
# Retrieve optional declarative OffloadPlan from the pipeline.
# When present, replaces heuristic block discovery.
plan = get_offload_plan(pipeline)
self._validate_component_offload_policy(modules)
if self.config.dlo_resident_layers and (plan is None or not plan.resident_dit_paths):
logger.warning(
@@ -1762,10 +1728,21 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
)
logger.info("DLO is using host tensors materialized by the ordinary loader")
# Apply the user policy to discovered auxiliary components. The default
# keeps the model-declared staged lifecycle; False leaves a component
# resident for repeated requests.
self._prepare_auxiliary_components(modules, plan)
# Keep VAE/encoders on CPU; move to GPU on-demand via hooks.
# This saves ~4.3 GB HBM per card (VAE 1.3 + encoder 1.1 + sound 1.9)
# during the DiT forward pass. They are only needed briefly for
# text-encoding (before DiT) and VAE-decode (after DiT).
for enc, enc_name in zip(modules.encoders, modules.encoder_names):
self._try_layerwise_offload_encoder(enc, enc_name, plan)
self._register_on_demand_hook(
enc, "encoder", stage_on_demand=plan is not None and enc_name in plan.on_demand_component_paths
)
for vae, vae_name in zip(modules.vaes, modules.vae_names):
self._register_on_demand_hook(
vae,
"vae",
stage_on_demand=(plan is not None and vae_name in plan.on_demand_component_paths),
)
# Move resident modules to GPU (small modules needed every forward)
for name, module in zip(modules.resident_names, modules.resident_modules):
+102 -10
View File
@@ -20,6 +20,17 @@ 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.
@@ -28,35 +39,100 @@ class BoundedAllocatorCache:
cached blocks are immediately reusable. This policy keeps the cache while
both of these bounds hold:
* cached-but-unallocated memory is at most 25% of device capacity; and
* 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
* at least 5% of device capacity is physically free.
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.
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.
"""
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)
return cached > int(total * self.max_cached_fraction) or free < int(total * self.min_free_fraction)
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
def release_if_needed(self, *, force: bool = False) -> bool:
"""Release cached blocks when a bound is crossed or release is forced."""
@@ -116,11 +192,12 @@ 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 = cache_retention
self.cache_retention = None
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:
@@ -198,8 +275,19 @@ 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:
@@ -239,6 +327,10 @@ 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()
-1
View File
@@ -569,7 +569,6 @@ class OrchestratorArgs:
enable_distributed_layerwise_offload: bool = False
dlo_use_allgather: bool = True
dlo_resident_layers: int = 0
dlo_offload_components: dict[str, bool] = field(default_factory=dict)
host_weight_runtime_mode: str = "disabled"
host_weight_runtime_root: str | None = None
dlo_host_registration_limit_gib: float = 0.0
-3
View File
@@ -1059,9 +1059,6 @@ class AsyncOmniEngine:
"enable_distributed_layerwise_offload": kwargs.get("enable_distributed_layerwise_offload", False),
"dlo_use_allgather": kwargs.get("dlo_use_allgather", True),
"dlo_resident_layers": kwargs.get("dlo_resident_layers", 0),
"dlo_offload_components": (
{} if kwargs.get("dlo_offload_components") is None else kwargs["dlo_offload_components"]
),
"host_weight_runtime_mode": kwargs.get("host_weight_runtime_mode", "disabled"),
"host_weight_runtime_root": kwargs.get("host_weight_runtime_root"),
"dlo_host_registration_limit_gib": kwargs.get("dlo_host_registration_limit_gib", 0.0),
-12
View File
@@ -770,18 +770,6 @@ class OmniServeCommand(CLISubcommand):
"under the existing pinned-memory policy and fall back to bounded staging when unavailable."
),
)
omni_config_group.add_argument(
"--dlo-offload-components",
type=json.loads,
default=None,
metavar="JSON",
help=(
"Per auxiliary encoder/VAE DLO policy as a JSON object, for example "
"'{\"text_encoder\": false}'. False keeps the whole component "
"resident; true permits the model's existing offload behavior. "
"DiT offload is not controlled by this map."
),
)
# Video model parameters (e.g., Wan2.2) - engine-level
omni_config_group.add_argument(
"--boundary-ratio",