Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / case_matrix.py: 88%
177 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Precision-validation case matrix generation from manifest topology specs."""
17from __future__ import annotations
19import dataclasses
20import json
21from dataclasses import asdict, dataclass, field
22from typing import Any, Iterable, Mapping, Optional
24from hyper_parallel.models.adapter_spec import RecomputePolicy
25from hyper_parallel.models.validation_spec import TopologyConstraint
27_TOPOLOGY_DEFAULTS = {
28 "tp": 1,
29 "cp": 1,
30 "ep": 1,
31 "fsdp": 1,
32 "sequence_parallel": False,
33 "recompute": {"layer_count": 0},
34}
35_PARALLEL_TOPOLOGY_FIELDS = (
36 "tp",
37 "cp",
38 "ep",
39 "fsdp",
40 "sequence_parallel",
41)
44@dataclass(frozen=True)
45class ValidationCase:
46 """One resolved precision-validation topology and execution role."""
48 name: str
49 topology: dict[str, Any]
50 kind: str = "topology"
51 compare_to: str = "baseline"
52 metadata: dict[str, Any] = field(default_factory=dict)
54 def to_dict(self) -> dict[str, Any]:
55 """Return a JSON-serializable representation."""
56 return asdict(self)
59def _case_name(topology: Mapping[str, Any], prefix: str = "") -> str:
60 ordered = (*_PARALLEL_TOPOLOGY_FIELDS, "recompute")
61 pieces = []
62 for name in ordered:
63 if name not in topology:
64 continue
65 value = topology[name]
66 if name == "recompute" and isinstance(value, Mapping):
67 if "layer_count" in value:
68 value = f"count{value['layer_count']}"
69 elif "layer_indices" in value:
70 value = "indices" + "-".join(str(index) for index in value["layer_indices"])
71 pieces.append(f"{name}{value}")
72 name = "_".join(str(piece).replace("/", "-") for piece in pieces)
73 return f"{prefix}_{name}" if prefix else name
76def _validate_recompute_selection(selection: Any, path: str) -> dict[str, Any]:
77 """Validate one exact layer-count or layer-index recompute selection."""
78 if not isinstance(selection, Mapping):
79 raise ValueError(f"{path} must be a mapping")
80 unsupported = tuple(
81 name for name in selection if name not in ("layer_count", "layer_indices")
82 )
83 if unsupported:
84 raise ValueError(
85 f"{path} contains unsupported fields {unsupported}; supported fields: "
86 "layer_count, layer_indices"
87 )
88 if ("layer_count" in selection) == ("layer_indices" in selection):
89 raise ValueError(
90 f"{path} must contain exactly one of layer_count or layer_indices"
91 )
92 if "layer_count" in selection:
93 layer_count = selection["layer_count"]
94 if isinstance(layer_count, bool) or not isinstance(layer_count, int) or layer_count < 0:
95 raise ValueError(f"{path}.layer_count must be a non-negative integer")
96 return {"layer_count": layer_count}
98 layer_indices = selection["layer_indices"]
99 if not isinstance(layer_indices, list) or not layer_indices:
100 raise ValueError(
101 f"{path}.layer_indices must be a non-empty list of non-negative integers"
102 )
103 if any(
104 isinstance(index, bool) or not isinstance(index, int) or index < 0
105 for index in layer_indices
106 ):
107 raise ValueError(
108 f"{path}.layer_indices must contain only non-negative integers"
109 )
110 if len(set(layer_indices)) != len(layer_indices):
111 raise ValueError(f"{path}.layer_indices must not contain duplicates")
112 return {"layer_indices": list(layer_indices)}
115def _validate_topology(topology: Mapping[str, Any], path: str) -> dict[str, Any]:
116 """Reject fields that the standard Trainer launcher cannot consume."""
117 if not isinstance(topology, Mapping):
118 raise ValueError(f"{path} must be a mapping")
119 resolved = dict(topology)
120 unsupported = tuple(name for name in resolved if name not in _TOPOLOGY_DEFAULTS)
121 if unsupported:
122 supported = ", ".join(_TOPOLOGY_DEFAULTS)
123 raise ValueError(
124 f"{path} contains unsupported fields {unsupported}; supported fields: {supported}"
125 )
126 if "recompute" in resolved:
127 resolved["recompute"] = _validate_recompute_selection(
128 resolved["recompute"],
129 f"{path}.recompute",
130 )
131 return resolved
134def _acceptance_class(
135 baseline: Mapping[str, Any],
136 candidate: Mapping[str, Any],
137) -> str:
138 """Classify tolerances from the parallel topology actually launched."""
139 for field_name in _PARALLEL_TOPOLOGY_FIELDS:
140 default = _TOPOLOGY_DEFAULTS[field_name]
141 if baseline.get(field_name, default) != candidate.get(field_name, default):
142 return "cross_topology"
143 return "same_topology"
146def _axis_cases(
147 matrix: Mapping[str, Any],
148 baseline: Mapping[str, Any],
149 seen: set[str],
150) -> list[ValidationCase]:
151 """Generate one case per declared single-axis deviation."""
152 axes = matrix.get("axes") or {}
153 if not isinstance(axes, Mapping):
154 raise ValueError("matrix.axes must be a mapping")
155 cases = []
156 for axis, values in axes.items():
157 if axis not in _TOPOLOGY_DEFAULTS:
158 supported = ", ".join(_TOPOLOGY_DEFAULTS)
159 raise ValueError(
160 f"matrix.axes contains unsupported field {axis!r}; "
161 f"supported fields: {supported}"
162 )
163 if not isinstance(values, list) or not values:
164 raise ValueError(f"matrix.axes.{axis} must be a non-empty list")
165 for value in values:
166 topology = dict(baseline)
167 topology[axis] = value
168 key = json.dumps(topology, sort_keys=True)
169 if key in seen:
170 continue
171 seen.add(key)
172 cases.append(
173 ValidationCase(_case_name(topology, f"axis-{axis}"), topology)
174 )
175 return cases
178def _combined_cases(
179 matrix: Mapping[str, Any],
180 baseline: Mapping[str, Any],
181 seen: set[str],
182) -> list[ValidationCase]:
183 """Generate one case per declared multi-axis combination."""
184 cases = []
185 for index, combined in enumerate(matrix.get("combined") or ()):
186 if not isinstance(combined, Mapping):
187 raise ValueError("matrix.combined entries must be mappings")
188 topology = dict(baseline)
189 topology.update(_validate_topology(combined, f"matrix.combined[{index}]"))
190 key = json.dumps(topology, sort_keys=True)
191 if key in seen:
192 continue
193 seen.add(key)
194 cases.append(ValidationCase(_case_name(topology, f"combined-{index}"), topology))
195 return cases
198def _recompute_cases(
199 matrix: Mapping[str, Any],
200 baseline: Mapping[str, Any],
201 seen: set[str],
202) -> list[ValidationCase]:
203 """Generate one case per requested recompute layer selection."""
204 requested_selections = matrix.get(
205 "recompute_selections",
206 ({"layer_count": 0},),
207 )
208 if not isinstance(requested_selections, (list, tuple)) or not requested_selections:
209 raise ValueError("matrix.recompute_selections must be a non-empty list")
210 cases = []
211 for index, requested_selection in enumerate(requested_selections):
212 selection = _validate_recompute_selection(
213 requested_selection,
214 f"matrix.recompute_selections[{index}]",
215 )
216 topology = dict(baseline)
217 topology.update(
218 _validate_topology(
219 matrix.get("recompute_topology") or {},
220 "matrix.recompute_topology",
221 )
222 )
223 topology["recompute"] = selection
224 key = json.dumps(topology, sort_keys=True)
225 if key in seen:
226 continue
227 seen.add(key)
228 cases.append(
229 ValidationCase(_case_name(topology, "recompute"), topology)
230 )
231 return cases
234def _ep1_fallback_case(
235 cases: list[ValidationCase],
236 baseline: Mapping[str, Any],
237) -> Optional[ValidationCase]:
238 """Retain a single-device-EP reference when every case parallelizes experts."""
239 if int(baseline.get("ep", 1)) <= 1 or any(
240 int(case.topology.get("ep", 1)) == 1 for case in cases
241 ):
242 return None
243 topology = dict(baseline)
244 topology["ep"] = 1
245 return ValidationCase(_case_name(topology, "axis-ep"), topology)
248def _production_validate_cases(
249 matrix: Mapping[str, Any],
250 baseline: Mapping[str, Any],
251) -> list[ValidationCase]:
252 """Generate the production/validate placement pair when declared."""
253 production_validate = matrix.get("production_validate_pair")
254 if not production_validate:
255 return []
256 production_validate = _validate_topology(
257 production_validate,
258 "matrix.production_validate_pair",
259 )
260 cases = []
261 for mode in ("production", "validate"):
262 topology = dict(baseline)
263 topology.update(production_validate)
264 topology["validate_placement"] = mode == "validate"
265 cases.append(
266 ValidationCase(
267 _case_name(topology, mode),
268 topology,
269 kind="production_validate",
270 )
271 )
272 return cases
275def _resume_cases(
276 matrix: Mapping[str, Any],
277 baseline: Mapping[str, Any],
278) -> list[ValidationCase]:
279 """Generate the declared same/cross-topology resume cases."""
280 cases = []
281 if matrix.get("same_topology_resume"):
282 cases.append(
283 ValidationCase(
284 "same_topology_resume",
285 dict(baseline),
286 kind="resume",
287 metadata={"resume": "same_topology"},
288 )
289 )
290 if matrix.get("cross_topology_resume"):
291 topology = dict(baseline)
292 topology.update(
293 _validate_topology(
294 matrix["cross_topology_resume"],
295 "matrix.cross_topology_resume",
296 )
297 )
298 if _acceptance_class(baseline, topology) == "same_topology":
299 raise ValueError(
300 "matrix.cross_topology_resume must change at least one of "
301 f"{_PARALLEL_TOPOLOGY_FIELDS}. A recompute-only change is not a "
302 "cross-topology restore; same_topology_resume always uses the "
303 "baseline recompute policy, and a recompute-only resume matrix "
304 "is not supported"
305 )
306 cases.append(
307 ValidationCase(
308 "cross_topology_resume",
309 topology,
310 kind="resume",
311 metadata={
312 "resume": "cross_topology",
313 "prepare_topology": dict(baseline),
314 },
315 )
316 )
317 return cases
320def _finalize_cases(
321 cases: list[ValidationCase],
322 baseline: Mapping[str, Any],
323 topology_constraints: Iterable[TopologyConstraint],
324) -> tuple[ValidationCase, ...]:
325 """Apply topology constraints and classify each case acceptance profile."""
326 resolved_cases = []
327 for case in cases:
328 for constraint in topology_constraints:
329 if not constraint.predicate(case.topology):
330 raise ValueError(
331 f"case {case.name!r} violates topology constraint "
332 f"{constraint.name!r}: {constraint.remediation}"
333 )
334 metadata = dict(case.metadata)
335 metadata["acceptance"] = _acceptance_class(baseline, case.topology)
336 resolved_cases.append(dataclasses.replace(case, metadata=metadata))
337 return tuple(resolved_cases)
340def generate_validation_cases(
341 matrix: Mapping[str, Any],
342 recompute_policy: Optional[RecomputePolicy] = None,
343 topology_constraints: Iterable[TopologyConstraint] = (),
344) -> tuple[ValidationCase, ...]:
345 """Generate a minimum single-axis-plus-combined validation matrix."""
346 baseline = dict(matrix.get("baseline") or {})
347 if not baseline:
348 raise ValueError("matrix.baseline is required")
349 baseline = _validate_topology(baseline, "matrix.baseline")
350 cases = [ValidationCase("baseline", baseline, compare_to="baseline")]
351 seen = {json.dumps(baseline, sort_keys=True)}
352 cases.extend(_axis_cases(matrix, baseline, seen))
353 cases.extend(_combined_cases(matrix, baseline, seen))
354 if recompute_policy is not None:
355 cases.extend(_recompute_cases(matrix, baseline, seen))
356 ep1_case = _ep1_fallback_case(cases, baseline)
357 if ep1_case is not None:
358 cases.insert(1, ep1_case)
359 cases.extend(_production_validate_cases(matrix, baseline))
360 cases.extend(_resume_cases(matrix, baseline))
361 return _finalize_cases(cases, baseline, topology_constraints)
364__all__ = [
365 "ValidationCase",
366 "generate_validation_cases",
367]