Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / config_adapter / _constraint_checker.py: 94%
250 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-08-04 05:18 +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"""Constraint checker for auto parallel strategy search.
17Validates cross-field constraints on a :class:`NormalizedConfig` instance:
18divisibility checks, device count limits, pipeline stage consistency,
19and required-field presence.
20"""
22from typing import Dict, List, Optional
24from hyper_parallel.auto_parallel.config_adapter._normalized_config import (
25 NormalizedConfig,
26 ValidationError,
27)
30def _err(field_path: str, message: str) -> ValidationError:
31 """Create an error-severity validation error."""
32 return ValidationError(field_path=field_path, message=message, severity="error")
35def _warn(field_path: str, message: str) -> ValidationError:
36 """Create a warning-severity validation error."""
37 return ValidationError(field_path=field_path, message=message, severity="warning")
40def _check_divisibility(
41 numerator_name: str,
42 numerator: int,
43 denominator_name: str,
44 denominator: int,
45) -> Optional[ValidationError]:
46 """Check that numerator is divisible by denominator. Returns None if OK."""
47 if denominator <= 1:
48 return None
49 if numerator <= 0:
50 return None
51 if numerator % denominator != 0:
52 return _err(
53 numerator_name,
54 f"{numerator_name} ({numerator}) must be divisible by "
55 f"{denominator_name} ({denominator}), "
56 f"remainder is {numerator % denominator}",
57 )
58 return None
61def _get_fixed_value(constraint: Dict, dim_key: str) -> Optional[int]:
62 """Look up a fixed dimension value from constraint dict."""
63 fixed_map = {
64 "dp": constraint.get("fixed_dp_degree"),
65 "tp": constraint.get("fixed_tp_degree"),
66 "pp": constraint.get("fixed_pp_degree"),
67 "cp": constraint.get("fixed_cp_degree"),
68 "ep": constraint.get("fixed_ep_degree"),
69 }
70 return fixed_map.get(dim_key)
73def _resolve_candidates(search_space: Dict, dim_key: str,
74 constraint: Dict, dim_label: str) -> List[int]:
75 """Return the effective candidate list for a dimension,
76 respecting fixed values from constraints."""
77 fixed = _get_fixed_value(constraint, dim_label)
78 if fixed is not None and fixed > 0:
79 return [fixed]
80 return search_space.get(dim_key, [1])
83def _candidates_or_default(search_space: Dict, dim_key: str,
84 constraint: Dict, dim_label: str) -> List[int]:
85 """Return candidates; if empty, default to [1]."""
86 candidates = _resolve_candidates(search_space, dim_key, constraint, dim_label)
87 if not candidates:
88 candidates = [1]
89 return candidates
92def _total_cards(cluster_spec: Dict) -> int:
93 """Compute total devices (num_nodes * cards_per_node)."""
94 nodes = cluster_spec.get("num_nodes", 1)
95 cards_per_node = cluster_spec.get("cards_per_node", 8)
96 if nodes <= 0 or cards_per_node <= 0:
97 return 0
98 return nodes * cards_per_node
101def validate(config: NormalizedConfig) -> List[ValidationError]:
102 """Validate cross-field constraints on a normalized configuration.
104 Performs all checks listed in Issue 126 Section 5:
105 divisibility, device product limit, pipeline constraints,
106 batch size relationships, and required-field presence.
108 Args:
109 config: The normalized configuration to validate.
111 Returns:
112 List of :class:`ValidationError` objects. An empty list
113 means the configuration is valid.
115 Example:
116 >>> errors = validate(config)
117 >>> for err in errors:
118 ... print(f"[{err.severity}] {err.field_path}: {err.message}")
119 """
120 errors: List[ValidationError] = []
122 model = config.model_spec
123 cluster = config.cluster_spec
124 search = config.search_space
125 constraint = config.constraint
126 pp_cfg = config.pp_config
128 _check_required_fields(errors, model)
129 _check_batch_size_relationships(errors, search, constraint)
130 _check_tp_divisibility(errors, model, search, constraint)
131 _check_cp_divisibility(errors, model, search, constraint)
132 _check_ep_divisibility(errors, model, search, constraint)
133 _check_fixed_dims_vs_search_space(errors, search, constraint)
134 _check_pipeline_constraints(errors, model, pp_cfg)
135 _check_layer_offset(errors, model, pp_cfg)
136 _check_layer_recompute(errors, model, pp_cfg)
137 _check_device_product_limit(errors, search, constraint, cluster)
138 _check_memory_limit(errors, cluster, constraint)
139 _check_dense_model_ep_cp_warning(errors, model, search)
140 _check_fsdp_hsdp_device_product(errors, search, constraint, cluster)
142 return errors
145def validate_strict(config: NormalizedConfig) -> None:
146 """Validate and raise ``ValueError`` on any ``"error"`` severity issues.
148 Args:
149 config: The normalized configuration to validate.
151 Raises:
152 ValueError: If one or more ``"error"`` severity issues are found,
153 with all messages concatenated.
154 """
155 errors = validate(config)
156 fatal = [e for e in errors if e.severity == "error"]
157 if fatal:
158 lines = "\n".join(f" [{e.severity}] {e.field_path}: {e.message}" for e in fatal)
159 raise ValueError(f"Configuration validation failed with {len(fatal)} error(s):\n{lines}")
162def _check_required_fields(errors: List[ValidationError], model: Dict) -> None:
163 """Check required model fields (num_hidden_layers, hidden_size,
164 num_attention_heads, vocab_size) are present and > 0."""
165 required = [
166 ("model.num_hidden_layers", model.get("num_hidden_layers", 0), "num_hidden_layers must be > 0"),
167 ("model.hidden_size", model.get("hidden_size", 0), "hidden_size must be > 0"),
168 ("model.num_attention_heads", model.get("num_attention_heads", 0), "num_attention_heads must be > 0"),
169 ("model.vocab_size", model.get("vocab_size", 0), "vocab_size must be > 0"),
170 ]
171 for field_path, value, message in required:
172 if value <= 0:
173 errors.append(_err(field_path, message))
176def _check_batch_size_relationships(
177 errors: List[ValidationError],
178 search: Dict,
179 constraint: Dict,
180) -> None:
181 """Check that global_batch_size is divisible by micro_batch_num and dp.
183 In FSDP/HSDP scenarios the effective data-parallel degree is
184 ``dp_shard * dp_replicate``, so both components are validated.
185 """
186 gbs = constraint.get("global_batch_size", 0)
187 if gbs <= 0:
188 return
190 mbn_list = search.get("micro_batch_num", [1])
191 for mbn in mbn_list:
192 if mbn > 0 and gbs % mbn != 0:
193 errors.append(_err(
194 "constraint.global_batch_size",
195 f"global_batch_size ({gbs}) must be divisible by "
196 f"micro_batch_num ({mbn}), remainder is {gbs % mbn}",
197 ))
199 dp_repl = _candidates_or_default(search, "data_parallel_replicate_degree", constraint, "dp")
200 dp_shard = _candidates_or_default(search, "data_parallel_shard_degree", constraint, "fsdp")
201 for repl in dp_repl:
202 for shard in dp_shard:
203 effective_dp = max(1, repl) * max(1, shard)
204 if effective_dp > 1 and gbs % effective_dp != 0:
205 errors.append(_err(
206 "constraint.global_batch_size",
207 f"global_batch_size ({gbs}) must be divisible by "
208 f"effective DP ({repl}*{shard}={effective_dp}), "
209 f"remainder is {gbs % effective_dp}",
210 ))
213def _check_tp_divisibility(
214 errors: List[ValidationError],
215 model: Dict,
216 search: Dict,
217 constraint: Dict,
218) -> None:
219 """Check that hidden_size, num_attention_heads, intermediate_size are divisible by tp."""
220 tp_list = _candidates_or_default(search, "tensor_parallel_degree", constraint, "tp")
221 dim = model.get("hidden_size", 0)
222 n_heads = model.get("num_attention_heads", 0)
223 inter_dim = model.get("intermediate_size", 0)
225 for tp in tp_list:
226 if tp <= 1:
227 continue
228 if dim > 0:
229 err = _check_divisibility("model.hidden_size", dim, "tp", tp)
230 if err:
231 errors.append(err)
232 if n_heads > 0:
233 err = _check_divisibility("model.num_attention_heads", n_heads, "tp", tp)
234 if err:
235 errors.append(err)
236 if inter_dim > 0:
237 err = _check_divisibility("model.intermediate_size", inter_dim, "tp", tp)
238 if err:
239 errors.append(err)
242def _check_cp_divisibility(
243 errors: List[ValidationError],
244 model: Dict,
245 search: Dict,
246 constraint: Dict,
247) -> None:
248 """Check that max_position_embeddings is divisible by cp."""
249 cp_list = _candidates_or_default(search, "context_parallel_degree", constraint, "cp")
250 seq_len = model.get("max_position_embeddings", 0)
252 for cp in cp_list:
253 if cp <= 1:
254 continue
255 if seq_len > 0:
256 err = _check_divisibility("model.max_position_embeddings", seq_len, "cp", cp)
257 if err:
258 errors.append(err)
261def _check_ep_divisibility(
262 errors: List[ValidationError],
263 model: Dict,
264 search: Dict,
265 constraint: Dict,
266) -> None:
267 """Check that num_experts is divisible by ep."""
268 ep_list = _candidates_or_default(search, "expert_parallel_degree", constraint, "ep")
269 num_experts = model.get("num_experts", 0)
271 for ep in ep_list:
272 if ep <= 1:
273 continue
274 if num_experts > 0:
275 err = _check_divisibility("model.num_experts", num_experts, "ep", ep)
276 if err:
277 errors.append(err)
280def _check_fixed_dims_vs_search_space(
281 errors: List[ValidationError],
282 search: Dict,
283 constraint: Dict,
284) -> None:
285 """Check that fixed dimension values are within the candidate search space."""
286 dim_map = {
287 "dp": "data_parallel_replicate_degree",
288 "tp": "tensor_parallel_degree",
289 "pp": "pipeline_parallel_degree",
290 "cp": "context_parallel_degree",
291 "ep": "expert_parallel_degree",
292 }
294 for dim_label, search_key in dim_map.items():
295 fixed = _get_fixed_value(constraint, dim_label)
296 if fixed is None:
297 continue
298 candidates = search.get(search_key, [])
299 if candidates and fixed not in candidates:
300 errors.append(_err(
301 f"constraint.fixed_{dim_label}_degree",
302 f"Fixed {dim_label}_degree ({fixed}) is not in the "
303 f"search space {candidates}",
304 ))
307def _check_pipeline_constraints(
308 errors: List[ValidationError],
309 model: Dict,
310 pp_cfg: Dict,
311) -> None:
312 """Check pipeline stage count, num_hidden_layers, and stage_partition consistency."""
313 pp_degree_raw = pp_cfg.get("pp_degree", 1)
314 pp_values = pp_degree_raw if isinstance(pp_degree_raw, list) else [pp_degree_raw]
315 pp_values = [v for v in pp_values if v > 1]
316 if not pp_values:
317 return
319 n_layers = model.get("num_hidden_layers", 0)
320 if n_layers <= 0:
321 return
323 for pp_degree_val in pp_values:
324 if pp_degree_val > n_layers:
325 errors.append(_err(
326 "pp_config.pp_degree",
327 f"pp_degree ({pp_degree_val}) exceeds the number of splittable "
328 f"layers ({n_layers})",
329 ))
331 stage_partition = pp_cfg.get("stage_partition", [])
332 stage_mode = pp_cfg.get("stage_partition_mode", "uniform")
333 if stage_mode == "manual" and stage_partition:
334 # Validate against every pp_degree candidate.
335 for pp_for_stages in pp_values:
336 if len(stage_partition) != pp_for_stages:
337 errors.append(_err(
338 "pp_config.stage_partition",
339 f"stage_partition has {len(stage_partition)} stages, "
340 f"but pp_degree is {pp_for_stages}",
341 ))
342 all_layers: set = set()
343 for stage_layers in stage_partition:
344 all_layers.update(stage_layers)
345 expected = set(range(n_layers))
346 missing = expected - all_layers
347 extra = all_layers - expected
348 if missing:
349 errors.append(_err(
350 "pp_config.stage_partition",
351 f"stage_partition does not cover layers: {sorted(missing)}",
352 ))
353 if extra:
354 errors.append(_err(
355 "pp_config.stage_partition",
356 f"stage_partition references non-existent layers: {sorted(extra)}",
357 ))
360def _check_layer_offset(
361 errors: List[ValidationError],
362 model: Dict,
363 pp_cfg: Dict,
364) -> None:
365 """Check that layer_offset_range is valid and within num_hidden_layers bounds."""
366 offset_range = pp_cfg.get("layer_offset_range", (0, 0))
367 if not isinstance(offset_range, (tuple, list)):
368 return
369 lo_min, lo_max = offset_range[0], offset_range[1]
370 if lo_min == 0 and lo_max == 0:
371 return
373 n_layers = model.get("num_hidden_layers", 0)
374 if lo_min > lo_max:
375 errors.append(_err(
376 "pp_config.layer_offset_range",
377 f"layer_offset_range min ({lo_min}) must be <= max ({lo_max})",
378 ))
379 if n_layers > 0:
380 if abs(lo_min) >= n_layers or abs(lo_max) >= n_layers:
381 errors.append(_err(
382 "pp_config.layer_offset_range",
383 f"layer_offset_range ({lo_min}, {lo_max}) exceeds "
384 f"num_layers ({n_layers})",
385 ))
388def _check_layer_recompute(
389 errors: List[ValidationError],
390 model: Dict,
391 pp_cfg: Dict,
392) -> None:
393 """Check that layer_recompute_layers indices are within [0, num_hidden_layers)."""
394 recompute_layers = pp_cfg.get("layer_recompute_layers", [])
395 if not recompute_layers:
396 return
398 n_layers = model.get("num_hidden_layers", 0)
399 if n_layers <= 0:
400 return
402 invalid = [idx for idx in recompute_layers
403 if idx < 0 or idx >= n_layers]
404 if invalid:
405 errors.append(_err(
406 "pp_config.layer_recompute_layers",
407 f"layer_recompute_layers references non-existent layers: {invalid}. "
408 f"Valid range: [0, {n_layers - 1}]",
409 ))
412def _check_device_product_limit(
413 errors: List[ValidationError],
414 search: Dict,
415 constraint: Dict,
416 cluster: Dict,
417) -> None:
418 """Check that parallel dimension product does not exceed available devices."""
419 total_devices = _total_cards(cluster)
420 if total_devices <= 0:
421 return
423 # DP is decomposed into replicate * shard when FSDP/HSDP is used.
424 # Compute the minimum product across all combinations to correctly
425 # check whether any valid DP decomposition fits the device budget.
426 dp_repl_vals = search.get("data_parallel_replicate_degree", [1]) or [1]
427 dp_shard_vals = search.get("data_parallel_shard_degree", [1]) or [1]
428 fixed_dp = constraint.get("fixed_dp_degree")
429 if fixed_dp is not None and fixed_dp > 0:
430 dp_min = fixed_dp
431 else:
432 dp_min = min(dp_repl_vals) * min(dp_shard_vals)
434 dim_keys = {
435 "tensor_parallel_degree": "tp",
436 "pipeline_parallel_degree": "pp",
437 "context_parallel_degree": "cp",
438 "expert_parallel_degree": "ep",
439 }
440 fixed_overrides = {
441 "tp": constraint.get("fixed_tp_degree"),
442 "pp": constraint.get("fixed_pp_degree"),
443 "cp": constraint.get("fixed_cp_degree"),
444 "ep": constraint.get("fixed_ep_degree"),
445 }
447 # Use the *minimum* product to check that at least one valid
448 # combination fits within the device budget. The enumerator
449 # (the strategy enumerator) will filter invalid combos.
450 min_product = dp_min
451 for search_key, dim_label in dim_keys.items():
452 fixed_val = fixed_overrides.get(dim_label)
453 if fixed_val is not None and fixed_val > 0:
454 min_product *= fixed_val
455 else:
456 candidates = search.get(search_key, [1])
457 if not candidates:
458 candidates = [1]
459 min_product *= min(candidates)
461 if min_product > total_devices:
462 errors.append(_err(
463 "search_space",
464 f"Minimum product of parallel dimensions ({min_product}) exceeds "
465 f"total available devices ({total_devices})",
466 ))
469def _check_memory_limit(
470 errors: List[ValidationError],
471 cluster: Dict,
472 constraint: Dict,
473) -> None:
474 """Check that memory_limit_gb is non-negative and does not exceed device memory."""
475 memory_limit = constraint.get("memory_limit_gb", 0.0)
476 if memory_limit < 0:
477 errors.append(_err(
478 "constraint.memory_limit_gb",
479 f"memory_limit_gb must be >= 0, got {memory_limit}",
480 ))
482 device_memory = cluster.get("device_memory_gb", 0.0)
483 if memory_limit > 0 and device_memory > 0:
484 if memory_limit > device_memory:
485 errors.append(_warn(
486 "constraint.memory_limit_gb",
487 f"memory_limit_gb ({memory_limit}) exceeds "
488 f"device_memory_gb ({device_memory})",
489 ))
492def _check_dense_model_ep_cp_warning(
493 errors: List[ValidationError],
494 model: Dict,
495 search: Dict,
496) -> None:
497 """Issue 126 Section 5 rule 9: warn when EP/CP are enabled on Dense models.
499 If ``moe_enabled`` is ``False`` and ``ep_degree`` or ``cp_degree``
500 candidates contain values > 1, emit a warning since EP/CP are
501 typically used for MoE and long-sequence scenarios respectively.
502 """
503 moe_enabled = model.get("moe_enabled", False)
504 if moe_enabled:
505 return
507 ep_vals = search.get("expert_parallel_degree", [1])
508 if ep_vals and any(v > 1 for v in ep_vals):
509 errors.append(_warn(
510 "search_space.expert_parallel_degree",
511 "Expert Parallelism (ep > 1) is configured but moe_enabled is "
512 "False. EP has no effect on Dense LLMs.",
513 ))
515 cp_vals = search.get("context_parallel_degree", [1])
516 if cp_vals and any(v > 1 for v in cp_vals):
517 errors.append(_warn(
518 "search_space.context_parallel_degree",
519 "Context Parallelism (cp > 1) is configured on a Dense LLM. "
520 "Ensure this is intentional for long-sequence scenarios.",
521 ))
524def _check_fsdp_hsdp_device_product(
525 errors: List[ValidationError],
526 search: Dict,
527 constraint: Dict,
528 cluster: Dict,
529) -> None:
530 """Issue 127 constraint: FSDP/HSDP device product validation.
532 Verifies that the sum of shard-degree and replicate-degree
533 (which together form a complete DP decomposition) does not
534 exceed available devices when combined with TP/PP/CP/EP.
535 """
536 total_devices = _total_cards(cluster)
537 if total_devices <= 0:
538 return
540 fixed_dp = constraint.get("fixed_dp_degree")
541 dp_shard_vals = search.get("data_parallel_shard_degree", [1])
542 dp_repl_vals = search.get("data_parallel_replicate_degree", [1])
544 if fixed_dp is not None and fixed_dp > 0:
545 dp_shard_vals = [fixed_dp]
546 dp_repl_vals = [1]
548 fixed_overrides = {
549 "tp": constraint.get("fixed_tp_degree"),
550 "pp": constraint.get("fixed_pp_degree"),
551 "cp": constraint.get("fixed_cp_degree"),
552 "ep": constraint.get("fixed_ep_degree"),
553 }
555 dim_keys = {
556 "tensor_parallel_degree": "tp",
557 "pipeline_parallel_degree": "pp",
558 "context_parallel_degree": "cp",
559 "expert_parallel_degree": "ep",
560 }
562 has_over_product = True
563 for dp_shard in (dp_shard_vals or [1]):
564 for dp_repl in (dp_repl_vals or [1]):
565 product = dp_shard * dp_repl
566 for search_key, dim_label in dim_keys.items():
567 fixed_val = fixed_overrides.get(dim_label)
568 if fixed_val is not None and fixed_val > 0:
569 product *= fixed_val
570 else:
571 candidates = search.get(search_key, [1]) or [1]
572 product *= min(candidates)
573 if product <= total_devices:
574 has_over_product = False
575 break
576 if not has_over_product:
577 break
579 if has_over_product:
580 errors.append(_err(
581 "search_space",
582 f"No FSDP/HSDP decomposition (dp_shard * dp_replicate) "
583 f"fits within total available devices ({total_devices}) "
584 f"when combined with TP/PP/CP/EP dimensions.",
585 ))