Skip to content

Commit 37712e2

Browse files
fix(checkpoint): save tuned modules nested in pipeline components
`save` stored only pipeline attributes that are tuned modules. A tuned module inside a pipeline component was left out, and save failed with "No tuned modules found". Each tuned module is now stored under its dotted path, e.g. `vae.decoder`. Modules that are not tuned are not saved. A module that the pipeline holds under several paths is saved once and restored as one module. Older checkpoints with one entry per alias load the module once. Ported from ai-dynamo/aitune branch devin/1792595450-checkpoint-nested-pipeline-wrappers. A missing path is reported as not found, a path to a non-module attribute as not a module. Co-authored-by: Konrad Nowicki <knowicki@nvidia.com> Signed-off-by: Łukasz Ślusarczyk <lukasz.slusarczyk@intel.com>
1 parent 6ef9b11 commit 37712e2

3 files changed

Lines changed: 203 additions & 30 deletions

File tree

‎ChangeLog.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,10 @@ SPDX-License-Identifier: Apache-2.0
77

88
## Unreleased
99

10+
- fix: `save` stores tuned modules nested in pipeline components under their dotted path, e.g. `vae.decoder`. Modules that are not tuned are not saved.
11+
12+
- fix: a tuned module that a pipeline holds under several paths is saved once and restored as one module.
13+
1014
- feat (breaking): `inspect` measures module depth in executed modules.
1115
Containers such as `nn.ModuleList` and modules that never run no longer add a level.
1216
`min_depth` is renamed to `depth`.

‎tests/unit/torch/checkpoint/test_torch_checkpoint.py‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,9 @@ def _tune_save_load_helper(model_factory, samples, checkpoint_name, wrap_model_f
7373
checkpoint (TorchCheckpoint): Checkpoint used to save and load the model.
7474
device_map (dict | None): Device map passed to `checkpoint.load`.
7575
76+
Returns:
77+
The loaded model.
78+
7679
Raises:
7780
AssertionError: If model predictions after load don't match expected results.
7881
"""
@@ -92,6 +95,7 @@ def _tune_save_load_helper(model_factory, samples, checkpoint_name, wrap_model_f
9295
with torch.no_grad():
9396
preds = model_loaded(test_data)
9497
torch.testing.assert_close(preds, expected, rtol=1e-3, atol=1e-3)
98+
return model_loaded
9599

96100

97101
def _wrap_whole_model(model, backend=None):
@@ -104,6 +108,16 @@ def _wrap_linear_layers(model):
104108
return model
105109

106110

111+
def _wrap_net_layers(pipeline):
112+
_wrap_linear_layers(pipeline.net)
113+
return pipeline
114+
115+
116+
def _alias_net_linear1(pipeline):
117+
pipeline.linear1 = pipeline.net.linear1
118+
return pipeline
119+
120+
107121
@requires_xpu
108122
@pytest.mark.parametrize(
109123
"backend",
@@ -227,6 +241,113 @@ def test_tune_save_load_complex_pipeline_layers(
227241
torch.testing.assert_close(preds, expected, rtol=1e-4, atol=1e-4)
228242

229243

244+
@requires_xpu
245+
@pytest.mark.parametrize(
246+
"device_map_keys",
247+
[
248+
pytest.param([], id="no_device_map"),
249+
pytest.param(["net.linear1", "net.linear2"], id="full_device_map"),
250+
pytest.param(["net.linear1"], id="partial_device_map"),
251+
],
252+
)
253+
def test_tune_save_load_pipeline_with_nested_tuned_modules(
254+
complex_pipeline_factory, sample, torch_device, checkpoint, device_map_keys
255+
):
256+
"""Test tuning, saving and loading a pipeline whose attribute contains tuned modules."""
257+
_tune_save_load_helper(
258+
complex_pipeline_factory,
259+
sample,
260+
"nested_pipeline_test.tt",
261+
_wrap_net_layers,
262+
checkpoint,
263+
device_map=dict.fromkeys(device_map_keys, torch_device),
264+
)
265+
266+
267+
@requires_xpu
268+
def test_tune_save_load_pipeline_with_nested_tuned_modules_rejects_leaf_name_in_device_map(
269+
complex_pipeline_factory, sample, torch_device, checkpoint
270+
):
271+
"""Test that a nested tuned module is addressed by its full path, not by its attribute name."""
272+
with expect_device_map_not_found():
273+
_tune_save_load_helper(
274+
complex_pipeline_factory,
275+
sample,
276+
"nested_pipeline_test.tt",
277+
_wrap_net_layers,
278+
checkpoint,
279+
device_map={"linear1": torch_device},
280+
)
281+
282+
283+
@requires_xpu
284+
def test_state_dict_from_pipeline_stores_only_nested_tuned_modules(complex_pipeline_factory, sample):
285+
"""Test that only the tuned module is saved, not the rest of its pipeline component."""
286+
pipeline = complex_pipeline_factory()
287+
pipeline.net.linear1 = Module(
288+
pipeline.net.linear1, "demo-simple1", strategy=OneBackendStrategy(TorchInductorBackend())
289+
)
290+
tune(pipeline, sample, batch_sizes=[1, 2])
291+
292+
state_dict = TorchCheckpoint.state_dict_from_pipeline(pipeline)
293+
294+
assert list(state_dict) == ["net.linear1"]
295+
assert list(state_dict["net.linear1"]) == [""]
296+
297+
298+
@requires_xpu
299+
def test_tune_save_load_pipeline_with_aliased_nested_tuned_module(complex_pipeline_factory, sample, checkpoint):
300+
"""Test that a tuned module reachable by two pipeline paths is saved once and restored as one module."""
301+
pipeline = _alias_net_linear1(_wrap_net_layers(complex_pipeline_factory()))
302+
tune(pipeline, sample, batch_sizes=[1, 2])
303+
assert list(TorchCheckpoint.state_dict_from_pipeline(pipeline)) == ["net.linear1", "net.linear2"]
304+
305+
pipeline = _tune_save_load_helper(
306+
lambda: _alias_net_linear1(complex_pipeline_factory()),
307+
sample,
308+
"aliased_pipeline_test.tt",
309+
lambda pipeline: _alias_net_linear1(_wrap_net_layers(pipeline)),
310+
checkpoint,
311+
)
312+
313+
assert isinstance(pipeline.net.linear1, Module)
314+
assert pipeline.linear1 is pipeline.net.linear1
315+
316+
317+
class _Vae(torch.nn.Module):
318+
def __init__(self):
319+
super().__init__()
320+
self.decoder = torch.nn.Linear(2, 2)
321+
322+
323+
class _PipelineWithDecoderAlias:
324+
def __init__(self):
325+
self.vae = _Vae()
326+
self.decoder = self.vae.decoder
327+
328+
329+
def test_load_state_dict_for_pipeline_restores_duplicate_alias_keys_once(monkeypatch):
330+
"""Test loading a checkpoint that stores an aliased module under each of its paths."""
331+
restored = _PipelineWithDecoderAlias()
332+
replacement = torch.nn.Linear(2, 2)
333+
calls = []
334+
335+
def load_module(module, state_dict, device_map, module_name):
336+
calls.append(module_name)
337+
device_map.pop(module_name, None)
338+
return replacement
339+
340+
monkeypatch.setattr(TorchCheckpoint, "load_state_dict_for_module", staticmethod(load_module))
341+
device_map = {"decoder": torch.device("cpu"), "vae.decoder": torch.device("cpu")}
342+
343+
TorchCheckpoint.load_state_dict_for_pipeline(restored, {"decoder": {}, "vae.decoder": {}}, device_map)
344+
345+
assert calls == ["decoder"]
346+
assert restored.decoder is replacement
347+
assert restored.vae.decoder is replacement
348+
assert device_map == {}
349+
350+
230351
@requires_xpu
231352
def test_get_pipeline_modules(pipeline_factory):
232353
pipeline = pipeline_factory()

‎torch_tweak/torch/checkpoint/torch_checkpoint.py‎

Lines changed: 78 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
# NOTE: This file has been modified by Intel Corporation.
55
"""Torch checkpoint module."""
66

7-
from collections import OrderedDict
7+
from collections import OrderedDict, defaultdict
88
from pathlib import Path
99
from typing import Any
1010

@@ -40,18 +40,16 @@ def save(
4040
4141
This method handles both individual torch modules and pipelines containing modules.
4242
For individual modules, it directly uses the module's state_dict() method.
43-
For pipelines, it extracts state dictionaries from all tuned modules in the pipeline.
44-
45-
The pipeline is expected to be an object with attributes that are tuned modules (WrapperModule instances).
46-
Only tuned modules are saved in the state dictionary, as regular torch modules don't have tuning information.
47-
48-
If no tuned modules are found in the pipeline, a ValueError is raised.
43+
For a pipeline, it saves each tuned module (WrapperModule instance) under its dotted path,
44+
e.g. `"vae.decoder"`. Modules that are not tuned are not saved.
4945
5046
Args:
5147
module_or_pipeline: The module or pipeline to save. This can be either a torch.nn.Module
52-
or a pipeline object with module attributes. For pipelines, only tuned modules
53-
(WrapperModule instances) will be saved.
48+
or a pipeline object with module attributes.
5449
path: The path where the module state should be saved.
50+
51+
Raises:
52+
ValueError: If the pipeline contains no tuned modules.
5553
"""
5654
if isinstance(module_or_pipeline, TorchModule):
5755
state_dict = module_or_pipeline.state_dict()
@@ -107,39 +105,55 @@ def load(
107105
return module_or_pipeline
108106

109107
@staticmethod
110-
def get_pipeline_modules(pipeline: Any) -> dict[str, torch.nn.Module]:
108+
def get_pipeline_modules(pipeline: Any) -> dict[str, TorchModule | WrapperModule]:
111109
"""Get the modules from the pipeline.
112110
113111
Args:
114112
pipeline: The pipeline object to extract modules from.
115113
116114
Returns:
117-
dict: A dictionary mapping attribute names to Module objects.
115+
dict: A dictionary mapping attribute names to modules. Tuned modules nested in a module
116+
attribute are added under their dotted path, e.g. `"vae.decoder"`.
118117
"""
119-
modules = {}
118+
modules: dict[str, TorchModule | WrapperModule] = {}
119+
120+
def add_wrapped_children(module: TorchModule, prefix: str) -> None:
121+
# `named_modules()` cannot be used: a wrapper forwards it to the wrapped module, which yields itself.
122+
for child_name, child in module.named_children():
123+
child_path = f"{prefix}.{child_name}"
124+
if isinstance(child, WrapperModule):
125+
modules[child_path] = child
126+
else:
127+
add_wrapped_children(child, child_path)
120128

121129
for name, attr in vars(pipeline).items():
122-
if isinstance(attr, (torch.nn.Module, WrapperModule)):
130+
if isinstance(attr, WrapperModule):
123131
modules[name] = attr
132+
elif isinstance(attr, TorchModule):
133+
modules[name] = attr
134+
add_wrapped_children(attr, name)
124135

125136
return modules
126137

127138
@staticmethod
128139
def state_dict_from_pipeline(pipeline: Any) -> dict[str, Any]:
129140
"""Extract the state_dict from a pipeline (e.g. HF Diffusers pipeline).
130141
142+
Each tuned module is stored once under its dotted path, also when the pipeline reaches it by several paths.
143+
131144
Args:
132145
pipeline: The pipeline to extract the state_dict from.
133146
134147
Returns:
135148
dict: The state_dict from the pipeline.
136149
"""
137150
state_dict = OrderedDict()
138-
pipeline_modules = TorchCheckpoint.get_pipeline_modules(pipeline)
151+
saved_module_ids = set()
139152

140-
for name, module in pipeline_modules.items():
141-
if isinstance(module, WrapperModule):
153+
for name, module in TorchCheckpoint.get_pipeline_modules(pipeline).items():
154+
if isinstance(module, WrapperModule) and id(module) not in saved_module_ids:
142155
state_dict[name] = module.state_dict()
156+
saved_module_ids.add(id(module))
143157
if not state_dict:
144158
raise ValueError("No tuned modules found in the pipeline")
145159
return state_dict
@@ -249,13 +263,35 @@ def replace_tuned_module(
249263
final_module.load_state_dict(state_dict, strict=False)
250264
return final_module
251265

266+
@staticmethod
267+
def get_pipeline_module_paths(pipeline: Any) -> dict[int, list[tuple[Any, str]]]:
268+
"""Get every place where the pipeline holds a module.
269+
270+
Args:
271+
pipeline: The pipeline to search.
272+
273+
Returns:
274+
dict: Maps `id()` of a module to the `(owner, attribute name)` pairs that hold it.
275+
"""
276+
module_paths: dict[int, list[tuple[Any, str]]] = defaultdict(list)
277+
278+
for name, module in vars(pipeline).items():
279+
if isinstance(module, (TorchModule, WrapperModule)):
280+
module_paths[id(module)].append((pipeline, name))
281+
if isinstance(module, TorchModule):
282+
for _, parent in module.named_modules(remove_duplicate=False):
283+
for child_name, child in parent.named_children():
284+
module_paths[id(child)].append((parent, child_name))
285+
286+
return module_paths
287+
252288
@staticmethod
253289
def load_state_dict_for_pipeline(pipeline, state_dict, device_map: dict[str, torch.device]):
254290
"""Load a state dictionary into a pipeline.
255291
256-
This method loads the state dictionary into the pipeline by matching module names
257-
in the state dictionary with attributes in the pipeline. Each matched module is
258-
replaced with the tuned module.
292+
This method loads the state dictionary into the pipeline by matching dotted module paths
293+
in the state dictionary with modules in the pipeline. Each matched module is replaced
294+
with the tuned module in every place where the pipeline holds it.
259295
260296
Args:
261297
pipeline: The pipeline to load the state dictionary into.
@@ -270,18 +306,30 @@ def load_state_dict_for_pipeline(pipeline, state_dict, device_map: dict[str, tor
270306
if a matched attribute is not a torch.nn.Module or WrapperModule.
271307
"""
272308
matched_modules = set()
309+
module_paths = TorchCheckpoint.get_pipeline_module_paths(pipeline)
310+
restored_module_ids = set()
273311
for name, module_state_dict in state_dict.items():
274-
if hasattr(pipeline, name):
275-
module = getattr(pipeline, name)
276-
if isinstance(module, (torch.nn.Module, WrapperModule)):
277-
matched_modules.add(name)
278-
setattr(
279-
pipeline,
280-
name,
281-
TorchCheckpoint.load_state_dict_for_module(module, module_state_dict, device_map, name),
282-
)
283-
else:
284-
raise ValueError(f"Module {name} is not a torch.nn.Module or WrapperModule")
312+
*parent_path, attribute = name.split(".")
313+
parent = pipeline
314+
for component in parent_path:
315+
parent = getattr(parent, component, None)
316+
if parent is None or not hasattr(parent, attribute):
317+
continue
318+
319+
module = getattr(parent, attribute)
320+
if not isinstance(module, (torch.nn.Module, WrapperModule)):
321+
raise ValueError(f"Module {name} is not a torch.nn.Module or WrapperModule")
322+
323+
matched_modules.add(name)
324+
if id(module) in restored_module_ids:
325+
# Older checkpoints store an aliased module under each of its paths.
326+
device_map.pop(name, None)
327+
continue
328+
329+
restored_module = TorchCheckpoint.load_state_dict_for_module(module, module_state_dict, device_map, name)
330+
for owner, child_name in module_paths[id(module)]:
331+
setattr(owner, child_name, restored_module)
332+
restored_module_ids.add(id(restored_module))
285333

286334
unmatched_keys = set(state_dict.keys()).difference(matched_modules)
287335
if unmatched_keys:

0 commit comments

Comments
 (0)