Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
729ccd7f11 |
No files matched your search
@@ -67,26 +67,6 @@ views and mapped ranges. A separate transport still decides whether to use
|
||||
registered mmap, private pinned staging, synchronous copies, or asynchronous
|
||||
H2D transfer.
|
||||
|
||||
## Payload validation at startup
|
||||
|
||||
Eligible diffusion loaders accept `--host-weight-runtime-validation` (or the
|
||||
`host_weight_runtime_validation` offline/stage configuration field):
|
||||
|
||||
- `manifest_and_metadata` is the existing default. It validates identity,
|
||||
metadata, tensor structure, and sizes, but does not detect payload-only
|
||||
corruption.
|
||||
- `full_checksum` reads and hashes every payload on warm acquisition before
|
||||
restoring weights. This adds startup I/O/CPU work proportional to artifact
|
||||
bytes; it is not a per-inference check.
|
||||
|
||||
For example, add `--host-weight-runtime-validation full_checksum` to an enabled
|
||||
HWR deployment. Preferred mode falls back to canonical loading and can publish
|
||||
a replacement after detecting corruption; required mode fails startup. Readers
|
||||
can use different validation levels in the same domain without changing artifact
|
||||
identity. Neither level authenticates an untrusted writer or prevents mutation
|
||||
after validation. Checksumming does not imply automatic recovery of a running
|
||||
service.
|
||||
|
||||
## Resolution behavior
|
||||
|
||||
The loader resolves the immutable canonical source and computes the exact
|
||||
|
||||
@@ -289,6 +289,13 @@ store, tensor ownership, or H2D payload. On success, each tensor view copies
|
||||
directly into the existing rotating HBM block buffers and the two private host
|
||||
staging slots are not allocated.
|
||||
|
||||
`--dlo-host-registration-mode` defaults to `auto`, preserving this behavior.
|
||||
Select `disabled` to bypass HWR mapping registration and use bounded host staging
|
||||
regardless of mapping size or budget. This leaves the existing pin-memory policy
|
||||
for staging buffers unchanged. The equivalent offline/stage configuration field
|
||||
is `dlo_host_registration_mode`; an explicit disable requires enabled HWR with
|
||||
no-AllGather DLO.
|
||||
|
||||
`--dlo-host-registration-limit-gib` is an optional per-worker preflight ceiling
|
||||
over page-aligned registered bytes. Zero adds no ceiling. A disabled pinned
|
||||
memory policy, unsupported platform/capability, over-budget mapping, or fully
|
||||
@@ -296,6 +303,14 @@ rolled-back registration error selects the existing two-slot staging path. A
|
||||
partial registration that cannot be rolled back aborts startup because closing
|
||||
the lease would unmap memory still owned by the platform.
|
||||
|
||||
Automatic registration logs its entry with device and budget, then each CUDA
|
||||
region's ordinal/total and byte size before calling `cudaHostRegister`. A
|
||||
completion message follows each successful call. These identify the last
|
||||
entered boundary if progress stops; they do not impose a timeout. A synchronous
|
||||
CUDA call cannot be safely cancelled by a Python timer, and a stuck process
|
||||
still requires external supervision. The explicit disable option avoids this
|
||||
registration path; it does not repair the underlying driver hang.
|
||||
|
||||
Direct checkpoint mmap remains unchanged and continues to use staging. It may
|
||||
require loader-owned per-block transforms, while the HWR artifact already
|
||||
contains final runtime bytes. DLO AllGather never receives an HWR final-layout
|
||||
|
||||
@@ -193,35 +193,37 @@ def test_resolve_execution_mode_rejects_unknown_execution_type():
|
||||
omni_config_module._resolve_execution_mode("unknown_execution_type")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("validation", ["manifest_and_metadata", "full_checksum"])
|
||||
def test_hwr_validation_survives_config_projection(validation: str):
|
||||
@pytest.mark.parametrize("registration_mode", ["auto", "disabled"])
|
||||
def test_hwr_registration_policy_reaches_offloader(tmp_path: Path, registration_mode: str):
|
||||
from vllm_omni.diffusion.data import OmniDiffusionConfig
|
||||
from vllm_omni.diffusion.offloader.base import OffloadConfig
|
||||
|
||||
override = StageDeployConfig(stage_id=0, host_weight_runtime_validation=validation)
|
||||
config = _from_pipeline_key(
|
||||
"dreamzero",
|
||||
deploy_config_path="dreamzero_tp1_cfg2",
|
||||
cli_overrides={"host_weight_runtime_validation": validation},
|
||||
)
|
||||
options = {
|
||||
"enable_distributed_layerwise_offload": True,
|
||||
"dlo_use_allgather": False,
|
||||
"host_weight_runtime_mode": "preferred",
|
||||
"host_weight_runtime_root": str(tmp_path / "unused-store"),
|
||||
"dlo_host_registration_mode": registration_mode,
|
||||
}
|
||||
config = _from_pipeline_key("dreamzero", deploy_config_path="dreamzero_tp1_cfg2", cli_overrides=options)
|
||||
stage = config.stage_by_id(0)
|
||||
assert isinstance(stage, VllmOmniDiffusionStageConfig)
|
||||
assert stage.diffusion_config.host_weight_runtime_validation == validation
|
||||
projection = omni_config_module._DiffusionConfigProjection.from_kwargs(
|
||||
host_weight_runtime_validation=override.host_weight_runtime_validation
|
||||
)
|
||||
assert projection.host_weight_runtime_validation == validation
|
||||
assert OmniDiffusionConfig(host_weight_runtime_validation=validation).host_weight_runtime_validation == validation
|
||||
assert omni_config_module._DiffusionConfigProjection().host_weight_runtime_validation == "manifest_and_metadata"
|
||||
assert OmniDiffusionConfig().host_weight_runtime_validation == "manifest_and_metadata"
|
||||
assert stage.diffusion_config.dlo_host_registration_mode == registration_mode
|
||||
offline = OmniDiffusionConfig(**options)
|
||||
transport = OffloadConfig.from_od_config(offline)
|
||||
assert transport.dlo_host_registration_mode == registration_mode
|
||||
assert transport.pin_cpu_memory
|
||||
assert transport.dlo_host_registration_limit_gib == 0.0
|
||||
assert not (tmp_path / "unused-store").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("validation", ["fs_verity", "unknown", None])
|
||||
def test_hwr_validation_rejects_unsupported_values(validation):
|
||||
@pytest.mark.parametrize("registration_mode", ["unknown", None, "disabled"])
|
||||
def test_hwr_registration_policy_rejects_invalid_or_ineligible_config(registration_mode):
|
||||
from vllm_omni.diffusion.data import OmniDiffusionConfig
|
||||
|
||||
for config_class in (OmniDiffusionConfig, omni_config_module._DiffusionConfigProjection):
|
||||
with pytest.raises(ValueError, match="host_weight_runtime_validation"):
|
||||
config_class(host_weight_runtime_validation=validation)
|
||||
with pytest.raises(ValueError, match="dlo_host_registration_mode"):
|
||||
config_class(dlo_host_registration_mode=registration_mode)
|
||||
|
||||
|
||||
def test_from_pipeline_config_preserves_current_pipeline_config_object():
|
||||
|
||||
@@ -260,78 +260,6 @@ def test_hwr_cold_publication_and_warm_restore_skip_ordinary_dit_loading(
|
||||
assert len(tuple((store_root / "source-digests-v1" / "entries").glob("*.json"))) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("validation", ["manifest_and_metadata", "full_checksum"])
|
||||
@pytest.mark.parametrize("mode", ["preferred", "required"])
|
||||
def test_hwr_loader_payload_integrity_policy(tmp_path: Path, monkeypatch, validation: str, mode: str):
|
||||
from vllm_omni.diffusion.offloader.startup import take_offload_startup_state
|
||||
|
||||
canonical = tmp_path / "canonical"
|
||||
canonical.mkdir()
|
||||
expected = torch.arange(4, dtype=torch.float32).to(torch.bfloat16).reshape(2, 2)
|
||||
save_file({"weight": expected}, str(canonical / "model.safetensors"))
|
||||
root = tmp_path / "store"
|
||||
|
||||
def make_loader(runtime_mode: str):
|
||||
config = _hwr_config(canonical, root, mode=runtime_mode)
|
||||
config.host_weight_runtime_validation = validation
|
||||
loader = DiffusersPipelineLoader(LoadConfig(), config)
|
||||
pipeline = _HWRPipeline(canonical)
|
||||
monkeypatch.setattr(loader, "_init_from_load_format", lambda *args, **kwargs: pipeline)
|
||||
return loader, pipeline
|
||||
|
||||
def close_plan(pipeline):
|
||||
state = take_offload_startup_state(pipeline)
|
||||
if state is not None and state.host_weight_plan is not None:
|
||||
carrier = state.host_weight_plan.lease_carrier
|
||||
if carrier is not None:
|
||||
carrier.close()
|
||||
|
||||
cold_loader, cold = make_loader("preferred")
|
||||
cold_loader.load_model(load_device="cpu", device=torch.device("cpu"))
|
||||
assert cold.load_count == 1
|
||||
close_plan(cold)
|
||||
payloads = list((root / "artifacts").glob("*/*.safetensors"))
|
||||
assert len(payloads) == 1
|
||||
payload = payloads[0]
|
||||
data = bytearray(payload.read_bytes())
|
||||
data[-1] ^= 1 # Keep the header and file length valid; change only a BF16 payload byte.
|
||||
original_mode = payload.stat().st_mode
|
||||
payload.chmod(0o600)
|
||||
payload.write_bytes(data)
|
||||
payload.chmod(original_mode)
|
||||
|
||||
loader, pipeline = make_loader(mode)
|
||||
try:
|
||||
if validation == "full_checksum" and mode == "required":
|
||||
with pytest.raises(RuntimeError, match="Host Weight Runtime resolution failed"):
|
||||
loader.load_model(load_device="cpu", device=torch.device("cpu"))
|
||||
assert pipeline.load_count == 0
|
||||
assert loader.take_host_weight_plan() is None
|
||||
else:
|
||||
loader.load_model(load_device="cpu", device=torch.device("cpu"))
|
||||
if validation == "manifest_and_metadata":
|
||||
assert pipeline.load_count == 0
|
||||
assert not torch.equal(pipeline.transformer.weight, expected)
|
||||
else:
|
||||
assert pipeline.load_count == 1
|
||||
assert torch.equal(pipeline.transformer.weight, expected)
|
||||
finally:
|
||||
close_plan(pipeline)
|
||||
|
||||
if validation == "full_checksum":
|
||||
if mode == "required":
|
||||
assert list((root / "deny").iterdir())
|
||||
else:
|
||||
assert list((root / "quarantine").iterdir())
|
||||
warm_loader, warm = make_loader(mode)
|
||||
try:
|
||||
warm_loader.load_model(load_device="cpu", device=torch.device("cpu"))
|
||||
assert warm.load_count == 0
|
||||
assert torch.equal(warm.transformer.weight, expected)
|
||||
finally:
|
||||
close_plan(warm)
|
||||
|
||||
|
||||
def test_maybe_fuse_distilled_lora_skips_when_hwr_warm_snapshot_present():
|
||||
cfg = SimpleNamespace(
|
||||
lora_backend="distill",
|
||||
|
||||
@@ -124,6 +124,33 @@ def test_registration_rejects_writable_or_over_budget_before_cuda(monkeypatch: p
|
||||
assert runtime.registered == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("second_result", [0, 7])
|
||||
def test_registration_logs_entry_before_each_runtime_call(monkeypatch: pytest.MonkeyPatch, second_result: int) -> None:
|
||||
runtime = _FakeRuntime([0, second_result])
|
||||
messages: list[str] = []
|
||||
monkeypatch.setattr(registration_module.logger, "info", lambda message, *args: messages.append(message % args))
|
||||
original_register = runtime.cudaHostRegister
|
||||
|
||||
def register(pointer: int, size: int, flags: int) -> int:
|
||||
ordinal = len(runtime.registered) + 1
|
||||
assert messages[-1] == f"Registering HWR host range {ordinal}/2: {size} bytes"
|
||||
return original_register(pointer, size, flags)
|
||||
|
||||
monkeypatch.setattr(runtime, "cudaHostRegister", register)
|
||||
monkeypatch.setattr(registration_module.torch.cuda, "cudart", lambda: runtime)
|
||||
monkeypatch.setattr(registration_module, "_consume_last_cuda_error", lambda _runtime, _error: None)
|
||||
regions = (_region("first", 0x1000, 4096), _region("second", 0x3000, 4096))
|
||||
if second_result:
|
||||
with pytest.raises(HostRegistrationError, match="error-7"):
|
||||
CudaHostRegistration.create(regions, max_bytes=None)
|
||||
assert runtime.unregistered == [0x1000]
|
||||
else:
|
||||
registration = CudaHostRegistration.create(regions, max_bytes=None)
|
||||
assert registration.close() == ()
|
||||
assert sum(message.startswith("Registering ") for message in messages) == 2
|
||||
assert sum(message.startswith("Registered ") for message in messages) == (1 if second_result else 2)
|
||||
|
||||
|
||||
def test_registration_requires_read_only_capability(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
runtime = _FakeRuntime([0], read_only_supported=False)
|
||||
monkeypatch.setattr(registration_module.torch.cuda, "cudart", lambda: runtime)
|
||||
|
||||
@@ -96,7 +96,7 @@ class TestDistributedLayerwiseOffloadHook:
|
||||
hook.offload_layer()
|
||||
assert not hook.is_materialized
|
||||
|
||||
def test_initialize_failure_keeps_next_block_materialized(self, monkeypatch):
|
||||
def test_initialize_failure_keeps_next_block_materialized(self, monkeypatch, patched_offload_runtime):
|
||||
current_block = nn.Linear(2, 2)
|
||||
next_block = nn.Linear(2, 2)
|
||||
expected = {name: tensor.detach().clone() for name, tensor in next_block.state_dict().items()}
|
||||
@@ -1373,10 +1373,14 @@ class TestMmapWeightLoading:
|
||||
|
||||
synchronize.assert_called()
|
||||
|
||||
def test_hwr_registration_failure_falls_back_to_bounded_staging(
|
||||
@pytest.mark.parametrize("registration_mode", ["auto", "disabled"])
|
||||
@pytest.mark.parametrize("limit_gib", [0.0, 1.5])
|
||||
def test_hwr_registration_selection_uses_bounded_staging(
|
||||
self,
|
||||
monkeypatch,
|
||||
patched_offload_runtime,
|
||||
registration_mode,
|
||||
limit_gib,
|
||||
):
|
||||
plan, carrier, lease = _fake_hwr_plan("hwr-registration-fallback")
|
||||
backend = DistributedLayerwiseOffloadBackend(
|
||||
@@ -1385,6 +1389,8 @@ class TestMmapWeightLoading:
|
||||
pin_cpu_memory=True,
|
||||
dp_size=1,
|
||||
dlo_use_allgather=False,
|
||||
dlo_host_registration_mode=registration_mode,
|
||||
dlo_host_registration_limit_gib=limit_gib,
|
||||
),
|
||||
torch.device("cpu"),
|
||||
host_weight_plan=plan,
|
||||
@@ -1394,12 +1400,14 @@ class TestMmapWeightLoading:
|
||||
|
||||
def fail_registration(*args, **kwargs):
|
||||
del args, kwargs
|
||||
assert registration_mode == "auto", "disabled mode entered platform registration"
|
||||
raise HostRegistrationError("registration unavailable in CPU test")
|
||||
|
||||
def allocate_unpinned_staging(hooks, resident_group=None):
|
||||
# Registration selection still observes pin_cpu_memory=True. The
|
||||
# CPU-only test avoids asking the host for CUDA-pinned allocations.
|
||||
for hook in hooks:
|
||||
assert hook.pin_memory, "registration policy disabled staging pinning"
|
||||
hook.pin_memory = False
|
||||
return original_staging_allocator(hooks, resident_group)
|
||||
|
||||
@@ -1412,6 +1420,7 @@ class TestMmapWeightLoading:
|
||||
assert backend._using_rank_local_mmap
|
||||
assert not backend._using_registered_mmap
|
||||
assert backend._host_registration is None
|
||||
assert backend.config.pin_cpu_memory
|
||||
assert all(len(hook.cpu_staging_buffers) == 2 for group in backend._all_hook_groups for hook in group)
|
||||
assert not lease.closed
|
||||
|
||||
@@ -2630,7 +2639,9 @@ class TestDistributedComponentSelection:
|
||||
assert registry is None or registry.get_hook("distributed_layerwise_offload") is None
|
||||
torch.testing.assert_close(block.weight, expected)
|
||||
|
||||
def test_multirank_enable_failure_cleanup_skips_restore_collective(self, monkeypatch, mocker):
|
||||
def test_multirank_enable_failure_cleanup_skips_restore_collective(
|
||||
self, monkeypatch, mocker, patched_offload_runtime
|
||||
):
|
||||
backend = DistributedLayerwiseOffloadBackend(
|
||||
OffloadConfig(
|
||||
strategy=OffloadStrategy.DISTRIBUTED_LAYER_WISE,
|
||||
@@ -2811,7 +2822,7 @@ class TestDistributedComponentSelection:
|
||||
with pytest.raises(ValueError, match="not declared replicated"):
|
||||
backend.enable(pipeline)
|
||||
|
||||
def test_encoder_allgather_rejects_stub_rank_before_block_discovery(self):
|
||||
def test_encoder_allgather_rejects_stub_rank_before_block_discovery(self, patched_offload_runtime):
|
||||
"""Every rank must reject an unsafe encoder group, including stub ranks."""
|
||||
backend = DistributedLayerwiseOffloadBackend(
|
||||
OffloadConfig(
|
||||
|
||||
@@ -503,12 +503,12 @@ def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():
|
||||
"--omni",
|
||||
"--enable-distributed-layerwise-offload",
|
||||
"--dlo-no-use-allgather",
|
||||
"--host-weight-runtime-validation",
|
||||
"full_checksum",
|
||||
"--host-weight-runtime-mode",
|
||||
"preferred",
|
||||
"--host-weight-runtime-root",
|
||||
"/var/cache/vllm-omni/hwr",
|
||||
"--dlo-host-registration-mode",
|
||||
"disabled",
|
||||
"--dlo-host-registration-limit-gib",
|
||||
"80",
|
||||
]
|
||||
@@ -521,11 +521,11 @@ def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():
|
||||
assert explicit_kwargs["host_weight_runtime_mode"] == "preferred"
|
||||
assert explicit_kwargs["host_weight_runtime_root"] == "/var/cache/vllm-omni/hwr"
|
||||
assert explicit_kwargs["dlo_host_registration_limit_gib"] == 80
|
||||
assert explicit_kwargs["host_weight_runtime_validation"] == "full_checksum"
|
||||
assert engine_args["host_weight_runtime_validation"] == "full_checksum"
|
||||
assert engine_args["host_weight_runtime_mode"] == "preferred"
|
||||
assert engine_args["host_weight_runtime_root"] == "/var/cache/vllm-omni/hwr"
|
||||
assert engine_args["dlo_host_registration_limit_gib"] == 80
|
||||
assert explicit_kwargs["dlo_host_registration_mode"] == "disabled"
|
||||
assert engine_args["dlo_host_registration_mode"] == "disabled"
|
||||
|
||||
|
||||
def test_serve_cli_accepts_diffusion_compile_controls():
|
||||
|
||||
@@ -739,7 +739,7 @@ class _DiffusionConfigProjection:
|
||||
dlo_resident_layers: int = Field(default=0, ge=0)
|
||||
host_weight_runtime_mode: Literal["disabled", "preferred", "required"] = "disabled"
|
||||
host_weight_runtime_root: str | None = None
|
||||
host_weight_runtime_validation: str = "manifest_and_metadata"
|
||||
dlo_host_registration_mode: str = "auto"
|
||||
dlo_host_registration_limit_gib: float = Field(default=0.0, ge=0)
|
||||
pin_cpu_memory: bool = True
|
||||
diffusion_compile_granularity: Literal["regional", "full"] = "regional"
|
||||
@@ -893,10 +893,10 @@ class _DiffusionConfigProjection:
|
||||
validate_host_weight_runtime_options(
|
||||
mode=self.host_weight_runtime_mode,
|
||||
root=self.host_weight_runtime_root,
|
||||
validation=self.host_weight_runtime_validation,
|
||||
)
|
||||
self.dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
|
||||
limit_gib=self.dlo_host_registration_limit_gib,
|
||||
mode=self.dlo_host_registration_mode,
|
||||
enable_dlo=self.enable_distributed_layerwise_offload,
|
||||
use_allgather=self.dlo_use_allgather,
|
||||
hwr_mode=self.host_weight_runtime_mode,
|
||||
|
||||
@@ -483,7 +483,7 @@ class StageDeployConfig:
|
||||
dlo_resident_layers: int | None = None
|
||||
host_weight_runtime_mode: str | None = None
|
||||
host_weight_runtime_root: str | None = None
|
||||
host_weight_runtime_validation: str | None = None
|
||||
dlo_host_registration_mode: str | None = None
|
||||
dlo_host_registration_limit_gib: float | None = None
|
||||
# Diffusion-specific debug and observability knobs.
|
||||
enable_diffusion_pipeline_profiler: bool | None = None
|
||||
|
||||
@@ -100,9 +100,7 @@ def normalize_omni_diffusion_kwargs(kwargs: Mapping[str, Any]) -> dict[str, Any]
|
||||
return normalized
|
||||
|
||||
|
||||
def validate_host_weight_runtime_options(
|
||||
*, mode: object, root: object, validation: object = "manifest_and_metadata"
|
||||
) -> None:
|
||||
def validate_host_weight_runtime_options(*, mode: object, root: object) -> None:
|
||||
"""Validate HWR policy without touching the configured storage domain.
|
||||
|
||||
Filesystem locality and store construction belong to the eligible loader
|
||||
@@ -111,8 +109,6 @@ def validate_host_weight_runtime_options(
|
||||
"""
|
||||
if mode not in {"disabled", "preferred", "required"}:
|
||||
raise ValueError("host_weight_runtime_mode must be disabled, preferred, or required")
|
||||
if validation not in ("manifest_and_metadata", "full_checksum"):
|
||||
raise ValueError("host_weight_runtime_validation must be manifest_and_metadata or full_checksum")
|
||||
if mode != "disabled" and (not isinstance(root, str) or not root.strip()):
|
||||
raise ValueError("enabled Host Weight Runtime requires host_weight_runtime_root")
|
||||
|
||||
@@ -123,8 +119,15 @@ def validate_dlo_host_registration_options(
|
||||
enable_dlo: bool,
|
||||
use_allgather: bool,
|
||||
hwr_mode: object,
|
||||
mode: object = "auto",
|
||||
) -> float:
|
||||
"""Validate the optional transport budget without probing CUDA or HWR."""
|
||||
if mode not in ("auto", "disabled"):
|
||||
raise ValueError("dlo_host_registration_mode must be auto or disabled")
|
||||
if mode == "disabled" and (not enable_dlo or use_allgather or hwr_mode == "disabled"):
|
||||
raise ValueError(
|
||||
"disabled dlo_host_registration_mode requires enabled no-AllGather DLO and Host Weight Runtime"
|
||||
)
|
||||
if not isinstance(limit_gib, (int, float, str)):
|
||||
raise TypeError(f"dlo_host_registration_limit_gib must be a number; got {type(limit_gib).__name__}")
|
||||
value = float(limit_gib)
|
||||
@@ -835,9 +838,9 @@ class OmniDiffusionConfig:
|
||||
# existing loader/storage path.
|
||||
host_weight_runtime_mode: str = "disabled"
|
||||
host_weight_runtime_root: str | None = None
|
||||
host_weight_runtime_validation: str = "manifest_and_metadata"
|
||||
# Optional per-worker ceiling for registering final-layout HWR mappings.
|
||||
# Zero adds no ceiling; pin_cpu_memory controls whether registration is tried.
|
||||
dlo_host_registration_mode: str = "auto"
|
||||
dlo_host_registration_limit_gib: float = 0.0
|
||||
|
||||
pin_cpu_memory: bool = True # Use pinned memory for faster transfers when offloading
|
||||
@@ -1246,10 +1249,10 @@ class OmniDiffusionConfig:
|
||||
validate_host_weight_runtime_options(
|
||||
mode=self.host_weight_runtime_mode,
|
||||
root=self.host_weight_runtime_root,
|
||||
validation=self.host_weight_runtime_validation,
|
||||
)
|
||||
self.dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
|
||||
limit_gib=self.dlo_host_registration_limit_gib,
|
||||
mode=self.dlo_host_registration_mode,
|
||||
enable_dlo=self.enable_distributed_layerwise_offload,
|
||||
use_allgather=self.dlo_use_allgather,
|
||||
hwr_mode=self.host_weight_runtime_mode,
|
||||
|
||||
@@ -26,7 +26,6 @@ from vllm_omni.diffusion.model_loader.host_weight_plan import HostWeightPlan, Te
|
||||
logger = init_logger(__name__)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from vllm_omni.diffusion.data import OmniDiffusionConfig
|
||||
from vllm_omni.diffusion.model_loader.host_weights.identity_adapter import FinalLayoutIdentityContext
|
||||
from vllm_omni.diffusion.model_loader.host_weights.source_identity import (
|
||||
NodeSourceDigestCache,
|
||||
@@ -41,8 +40,6 @@ class _HWRCommitError(RuntimeError):
|
||||
class HWRLoaderMixin:
|
||||
"""Optional final-layout HWR behavior shared by diffusion loaders."""
|
||||
|
||||
od_config: OmniDiffusionConfig
|
||||
|
||||
@staticmethod
|
||||
def _identity_value(value: object) -> object:
|
||||
"""Convert config objects into deterministic identity metadata."""
|
||||
@@ -322,12 +319,10 @@ class HWRLoaderMixin:
|
||||
HostWeightLeaseCarrier,
|
||||
HostWeightRuntime,
|
||||
HostWeightRuntimeConfig,
|
||||
IntegrityPolicy,
|
||||
ProductionPolicy,
|
||||
ResolutionOutcome,
|
||||
RuntimeMode,
|
||||
StorageDomainPolicy,
|
||||
ValidationLevel,
|
||||
)
|
||||
from vllm_omni.host_weight_runtime.filesystem import detect_storage_class
|
||||
|
||||
@@ -368,11 +363,6 @@ class HWRLoaderMixin:
|
||||
HostWeightRuntimeConfig(
|
||||
mode=mode,
|
||||
domain=StorageDomainPolicy(root=root, storage_class=detect_storage_class(root)),
|
||||
integrity=IntegrityPolicy(
|
||||
local_lookup=ValidationLevel(
|
||||
getattr(self.od_config, "host_weight_runtime_validation", "manifest_and_metadata")
|
||||
),
|
||||
),
|
||||
production=ProductionPolicy(
|
||||
allow_local_build=False,
|
||||
allow_post_load_publish=True,
|
||||
|
||||
@@ -73,12 +73,15 @@ class OffloadConfig:
|
||||
# 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
|
||||
dlo_host_registration_mode: str = "auto"
|
||||
# ``None`` preserves the model's legacy plan-driven component topology;
|
||||
# a frozenset is an explicit compact-API selection.
|
||||
components: frozenset[str] | None = None
|
||||
dlo_transfers: dict[str, DLOTransfer] | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.dlo_host_registration_mode not in ("auto", "disabled"):
|
||||
raise ValueError("dlo_host_registration_mode must be auto or disabled")
|
||||
if self.components is not None:
|
||||
self.components = parse_offload_components(self.components)
|
||||
if self.dlo_transfers is None:
|
||||
@@ -174,7 +177,9 @@ class OffloadConfig:
|
||||
dlo_transfers = dict(resolved.transfers)
|
||||
dlo_resident_layers = resolved.resident_layers
|
||||
dit_uses_allgather = resolved.uses_allgather(DIT_COMPONENT)
|
||||
registration_mode = getattr(od_config, "dlo_host_registration_mode", "auto")
|
||||
dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
|
||||
mode=registration_mode,
|
||||
limit_gib=getattr(od_config, "dlo_host_registration_limit_gib", 0.0),
|
||||
enable_dlo=enable_distributed_layerwise_offload,
|
||||
use_allgather=dit_uses_allgather,
|
||||
@@ -210,6 +215,7 @@ class OffloadConfig:
|
||||
dlo_use_allgather=dit_uses_allgather,
|
||||
dlo_resident_layers=dlo_resident_layers,
|
||||
dlo_host_registration_limit_gib=dlo_host_registration_limit_gib,
|
||||
dlo_host_registration_mode=registration_mode,
|
||||
components=components,
|
||||
dlo_transfers=dlo_transfers,
|
||||
)
|
||||
|
||||
@@ -6,12 +6,14 @@ from __future__ import annotations
|
||||
|
||||
import ctypes
|
||||
import mmap
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Protocol
|
||||
|
||||
import torch
|
||||
from vllm.logger import init_logger
|
||||
|
||||
from vllm_omni.host_weight_runtime import MappedHostRegion
|
||||
|
||||
@@ -21,6 +23,8 @@ from .host_registration import (
|
||||
HostRegistrationError,
|
||||
)
|
||||
|
||||
logger = init_logger(__name__)
|
||||
|
||||
_CUDA_HOST_REGISTER_READ_ONLY = 0x08
|
||||
_CUDA_DEVICE_ATTRIBUTE_HOST_REGISTER_READ_ONLY_SUPPORTED = 113
|
||||
|
||||
@@ -187,7 +191,9 @@ class CudaHostRegistration:
|
||||
|
||||
registered: list[_AddressRange] = []
|
||||
try:
|
||||
for region in mapped:
|
||||
for index, region in enumerate(mapped, start=1):
|
||||
logger.info("Registering HWR host range %d/%d: %d bytes", index, len(mapped), region.size)
|
||||
started = time.perf_counter()
|
||||
error = runtime.cudaHostRegister(region.start, region.size, _CUDA_HOST_REGISTER_READ_ONLY)
|
||||
if int(error) != 0:
|
||||
raise HostRegistrationError(
|
||||
@@ -195,6 +201,9 @@ class CudaHostRegistration:
|
||||
f"[{region.start:#x}, {region.end:#x}): {_handled_error_message(runtime, error)}"
|
||||
)
|
||||
registered.append(region)
|
||||
logger.info(
|
||||
"Registered HWR host range %d/%d in %.3f s", index, len(mapped), time.perf_counter() - started
|
||||
)
|
||||
except Exception as exc:
|
||||
rollback_errors: list[str] = []
|
||||
rollback_failed: list[_AddressRange] = []
|
||||
|
||||
@@ -1080,6 +1080,9 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
|
||||
lease = self._host_weight_lease
|
||||
if lease is None:
|
||||
return False
|
||||
if self.config.dlo_host_registration_mode == "disabled":
|
||||
logger.info("HWR mmap registration disabled by transport policy; using bounded host staging")
|
||||
return False
|
||||
if not self.config.pin_cpu_memory:
|
||||
logger.info("HWR mmap registration disabled by pin_cpu_memory=False; using bounded host staging")
|
||||
return False
|
||||
@@ -1089,6 +1092,12 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
|
||||
|
||||
limit_gib = self.config.dlo_host_registration_limit_gib
|
||||
max_bytes = int(limit_gib * 1024**3) if limit_gib > 0 else None
|
||||
logger.info(
|
||||
"Starting HWR mmap registration on %s: %d source range(s), budget_bytes=%s",
|
||||
self.device,
|
||||
len(lease.mapped_regions),
|
||||
max_bytes,
|
||||
)
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
registration = register_host_mappings(
|
||||
|
||||
@@ -582,7 +582,7 @@ class OrchestratorArgs:
|
||||
dlo_resident_layers: int = 0
|
||||
host_weight_runtime_mode: str = "disabled"
|
||||
host_weight_runtime_root: str | None = None
|
||||
host_weight_runtime_validation: str = "manifest_and_metadata"
|
||||
dlo_host_registration_mode: str = "auto"
|
||||
dlo_host_registration_limit_gib: float = 0.0
|
||||
boundary_ratio: float | None = None
|
||||
flow_shift: float | None = None
|
||||
|
||||
@@ -1145,7 +1145,7 @@ class AsyncOmniEngine:
|
||||
"dlo_resident_layers": kwargs.get("dlo_resident_layers", 0),
|
||||
"host_weight_runtime_mode": kwargs.get("host_weight_runtime_mode", "disabled"),
|
||||
"host_weight_runtime_root": kwargs.get("host_weight_runtime_root"),
|
||||
"host_weight_runtime_validation": kwargs.get("host_weight_runtime_validation", "manifest_and_metadata"),
|
||||
"dlo_host_registration_mode": kwargs.get("dlo_host_registration_mode", "auto"),
|
||||
"dlo_host_registration_limit_gib": kwargs.get("dlo_host_registration_limit_gib", 0.0),
|
||||
"enforce_eager": False if kwargs.get("enforce_eager") is None else kwargs.get("enforce_eager"),
|
||||
"diffusion_compile_granularity": (
|
||||
|
||||
@@ -770,15 +770,6 @@ class OmniServeCommand(CLISubcommand):
|
||||
"preferred first."
|
||||
),
|
||||
)
|
||||
omni_config_group.add_argument(
|
||||
"--host-weight-runtime-validation",
|
||||
choices=("manifest_and_metadata", "full_checksum"),
|
||||
default="manifest_and_metadata",
|
||||
help=(
|
||||
"HWR warm-acquisition validation. The default checks metadata, not payload corruption; "
|
||||
"full_checksum reads and hashes every payload before restoring weights."
|
||||
),
|
||||
)
|
||||
omni_config_group.add_argument(
|
||||
"--host-weight-runtime-root",
|
||||
type=str,
|
||||
@@ -789,6 +780,17 @@ class OmniServeCommand(CLISubcommand):
|
||||
"the same persistent path for population and serving."
|
||||
),
|
||||
)
|
||||
omni_config_group.add_argument(
|
||||
"--dlo-host-registration-mode",
|
||||
choices=("auto", "disabled"),
|
||||
default="auto",
|
||||
help=(
|
||||
"HWR transport registration policy for eligible no-AllGather DLO. "
|
||||
"auto attempts registered direct H2D under the existing pin-memory and budget policy; "
|
||||
"disabled selects bounded staging without registering the HWR mappings. "
|
||||
"This preserves the pin-memory policy for staging buffers."
|
||||
),
|
||||
)
|
||||
omni_config_group.add_argument(
|
||||
"--dlo-host-registration-limit-gib",
|
||||
type=float,
|
||||
|
||||
Reference in new issue
Block a user