Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e1c4b6153a | ||
|
|
33297bdb9d | ||
|
|
f4394f1d3d | ||
|
|
c9bb65902e | ||
|
|
b78db6036a | ||
|
|
27f9726a6a | ||
|
|
c5295f3ffe | ||
|
|
ba5943dddb |
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():
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user