Coverage for /home/jenkins/.local/lib/python3.10/site-packages/hyper_parallel/auto_parallel/config_adapter/_constraint

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

7# http://www.apache.org/licenses/LICENSE-2.0

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

100

101def validate(config: NormalizedConfig) -> List[ValidationError]:

102 """Validate cross-field constraints on a normalized configuration.

103

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.

107

108 Args:

109 config: The normalized configuration to validate.

110

111 Returns:

112 List of :class:`ValidationError` objects. An empty list

113 means the configuration is valid.

114

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] = []

121

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

127

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)

141

142 return errors

143

144

145def validate_strict(config: NormalizedConfig) -> None:

146 """Validate and raise ``ValueError`` on any ``"error"`` severity issues.

147

148 Args:

149 config: The normalized configuration to validate.

150

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}")

160

161

162def _check_required_fields(errors: List[ValidationError], model: Dict) -> None:

163 """Check that required model fields (n_layers, dim, n_heads, vocab_size) are present and > 0."""

164 required = [

165 ("model.n_layers", model.get("n_layers", 0), "n_layers must be > 0"),

166 ("model.dim", model.get("dim", 0), "dim must be > 0"),

167 ("model.n_heads", model.get("n_heads", 0), "n_heads must be > 0"),

168 ("model.vocab_size", model.get("vocab_size", 0), "vocab_size must be > 0"),

169 ]

170 for field_path, value, message in required:

171 if value <= 0:

172 errors.append(_err(field_path, message))

173

174

175def _check_batch_size_relationships(

176 errors: List[ValidationError],

177 search: Dict,

178 constraint: Dict,

179) -> None:

180 """Check that global_batch_size is divisible by micro_batch_num and dp.

181

182 In FSDP/HSDP scenarios the effective data-parallel degree is

183 ``dp_shard * dp_replicate``, so both components are validated.

184 """

185 gbs = constraint.get("global_batch_size", 0)

186 if gbs <= 0:

187 return

188

189 mbn_list = search.get("micro_batch_num", [1])

190 for mbn in mbn_list:

191 if mbn > 0 and gbs % mbn != 0:

192 errors.append(_err(

193 "constraint.global_batch_size",

194 f"global_batch_size ({gbs}) must be divisible by "

195 f"micro_batch_num ({mbn}), remainder is {gbs % mbn}",

196 ))

197

198 dp_repl = _candidates_or_default(search, "data_parallel_replicate_degree", constraint, "dp")

199 dp_shard = _candidates_or_default(search, "data_parallel_shard_degree", constraint, "fsdp")

200 for repl in dp_repl:

201 for shard in dp_shard:

202 effective_dp = max(1, repl) * max(1, shard)

203 if effective_dp > 1 and gbs % effective_dp != 0:

204 errors.append(_err(

205 "constraint.global_batch_size",

206 f"global_batch_size ({gbs}) must be divisible by "

207 f"effective DP ({repl}*{shard}={effective_dp}), "

208 f"remainder is {gbs % effective_dp}",

209 ))

210

211

212def _check_tp_divisibility(

213 errors: List[ValidationError],

214 model: Dict,

215 search: Dict,

216 constraint: Dict,

217) -> None:

218 """Check that dim, n_heads, inter_dim are divisible by tp."""

219 tp_list = _candidates_or_default(search, "tensor_parallel_degree", constraint, "tp")

220 dim = model.get("dim", 0)

221 n_heads = model.get("n_heads", 0)

222 inter_dim = model.get("inter_dim", 0)

223

224 for tp in tp_list:

225 if tp <= 1:

226 continue

227 if dim > 0:

228 err = _check_divisibility("model.dim", dim, "tp", tp)

229 if err:

230 errors.append(err)

231 if n_heads > 0:

232 err = _check_divisibility("model.n_heads", n_heads, "tp", tp)

233 if err:

234 errors.append(err)

235 if inter_dim > 0:

236 err = _check_divisibility("model.inter_dim", inter_dim, "tp", tp)

237 if err:

238 errors.append(err)

239

240

241def _check_cp_divisibility(

242 errors: List[ValidationError],

243 model: Dict,

244 search: Dict,

245 constraint: Dict,

246) -> None:

247 """Check that seq_len is divisible by cp."""

248 cp_list = _candidates_or_default(search, "context_parallel_degree", constraint, "cp")

249 seq_len = model.get("seq_len", 0)

250

251 for cp in cp_list:

252 if cp <= 1:

253 continue

254 if seq_len > 0:

255 err = _check_divisibility("model.seq_len", seq_len, "cp", cp)

256 if err:

257 errors.append(err)

258

259

260def _check_ep_divisibility(

261 errors: List[ValidationError],

262 model: Dict,

263 search: Dict,

264 constraint: Dict,

265) -> None:

266 """Check that num_experts is divisible by ep."""

267 ep_list = _candidates_or_default(search, "expert_parallel_degree", constraint, "ep")

268 num_experts = model.get("num_experts", 0)

269

270 for ep in ep_list:

271 if ep <= 1:

272 continue

273 if num_experts > 0:

274 err = _check_divisibility("model.num_experts", num_experts, "ep", ep)

275 if err:

276 errors.append(err)

277

278

279def _check_fixed_dims_vs_search_space(

280 errors: List[ValidationError],

281 search: Dict,

282 constraint: Dict,

283) -> None:

284 """Check that fixed dimension values are within the candidate search space."""

285 dim_map = {

286 "dp": "data_parallel_replicate_degree",

287 "tp": "tensor_parallel_degree",

288 "pp": "pipeline_parallel_degree",

289 "cp": "context_parallel_degree",

290 "ep": "expert_parallel_degree",

291 }

292

293 for dim_label, search_key in dim_map.items():

294 fixed = _get_fixed_value(constraint, dim_label)

295 if fixed is None:

296 continue

297 candidates = search.get(search_key, [])

298 if candidates and fixed not in candidates:

299 errors.append(_err(

300 f"constraint.fixed_{dim_label}_degree",

301 f"Fixed {dim_label}_degree ({fixed}) is not in the "

302 f"search space {candidates}",

303 ))

304

305

306def _check_pipeline_constraints(

307 errors: List[ValidationError],

308 model: Dict,

309 pp_cfg: Dict,

310) -> None:

311 """Check pipeline stage count, n_layers, and stage_partition consistency."""

312 pp_degree_raw = pp_cfg.get("pp_degree", 1)

313 pp_values = pp_degree_raw if isinstance(pp_degree_raw, list) else [pp_degree_raw]

314 pp_values = [v for v in pp_values if v > 1]

315 if not pp_values:

316 return

317

318 n_layers = model.get("n_layers", 0)

319 if n_layers <= 0:

320 return

321

322 for pp_degree_val in pp_values:

323 if pp_degree_val > n_layers:

324 errors.append(_err(

325 "pp_config.pp_degree",

326 f"pp_degree ({pp_degree_val}) exceeds the number of splittable "

327 f"layers ({n_layers})",

328 ))

329

330 stage_partition = pp_cfg.get("stage_partition", [])

331 stage_mode = pp_cfg.get("stage_partition_mode", "uniform")

332 if stage_mode == "manual" and stage_partition:

333 # Validate against every pp_degree candidate.

334 for pp_for_stages in pp_values:

335 if len(stage_partition) != pp_for_stages:

336 errors.append(_err(

337 "pp_config.stage_partition",

338 f"stage_partition has {len(stage_partition)} stages, "

339 f"but pp_degree is {pp_for_stages}",

340 ))

341 all_layers: set = set()

342 for stage_layers in stage_partition:

343 all_layers.update(stage_layers)

344 expected = set(range(n_layers))

345 missing = expected - all_layers

346 extra = all_layers - expected

347 if missing:

348 errors.append(_err(

349 "pp_config.stage_partition",

350 f"stage_partition does not cover layers: {sorted(missing)}",

351 ))

352 if extra:

353 errors.append(_err(

354 "pp_config.stage_partition",

355 f"stage_partition references non-existent layers: {sorted(extra)}",

356 ))

357

358

359def _check_layer_offset(

360 errors: List[ValidationError],

361 model: Dict,

362 pp_cfg: Dict,

363) -> None:

364 """Check that layer_offset_range is valid and within n_layers bounds."""

365 offset_range = pp_cfg.get("layer_offset_range", (0, 0))

366 if not isinstance(offset_range, (tuple, list)):

367 return

368 lo_min, lo_max = offset_range[0], offset_range[1]

369 if lo_min == 0 and lo_max == 0:

370 return

371

372 n_layers = model.get("n_layers", 0)

373 if lo_min > lo_max:

374 errors.append(_err(

375 "pp_config.layer_offset_range",

376 f"layer_offset_range min ({lo_min}) must be <= max ({lo_max})",

377 ))

378 if n_layers > 0:

379 if abs(lo_min) >= n_layers or abs(lo_max) >= n_layers:

380 errors.append(_err(

381 "pp_config.layer_offset_range",

382 f"layer_offset_range ({lo_min}, {lo_max}) exceeds "

383 f"num_layers ({n_layers})",

384 ))

385

386

387def _check_layer_recompute(

388 errors: List[ValidationError],

389 model: Dict,

390 pp_cfg: Dict,

391) -> None:

392 """Check that layer_recompute_layers indices are within [0, n_layers)."""

393 recompute_layers = pp_cfg.get("layer_recompute_layers", [])

394 if not recompute_layers:

395 return

396

397 n_layers = model.get("n_layers", 0)

398 if n_layers <= 0:

399 return

400

401 invalid = [idx for idx in recompute_layers

402 if idx < 0 or idx >= n_layers]

403 if invalid:

404 errors.append(_err(

405 "pp_config.layer_recompute_layers",

406 f"layer_recompute_layers references non-existent layers: {invalid}. "

407 f"Valid range: [0, {n_layers - 1}]",

408 ))

409

410

411def _check_device_product_limit(

412 errors: List[ValidationError],

413 search: Dict,

414 constraint: Dict,

415 cluster: Dict,

416) -> None:

417 """Check that parallel dimension product does not exceed available devices."""

418 total_devices = _total_cards(cluster)

419 if total_devices <= 0:

420 return

421

422 # DP is decomposed into replicate * shard when FSDP/HSDP is used.

423 # Compute the minimum product across all combinations to correctly

424 # check whether any valid DP decomposition fits the device budget.

425 dp_repl_vals = search.get("data_parallel_replicate_degree", [1]) or [1]

426 dp_shard_vals = search.get("data_parallel_shard_degree", [1]) or [1]

427 fixed_dp = constraint.get("fixed_dp_degree")

428 if fixed_dp is not None and fixed_dp > 0:

429 dp_min = fixed_dp

430 else:

431 dp_min = min(dp_repl_vals) * min(dp_shard_vals)

432

433 dim_keys = {

434 "tensor_parallel_degree": "tp",

435 "pipeline_parallel_degree": "pp",

436 "context_parallel_degree": "cp",

437 "expert_parallel_degree": "ep",

438 }

439 fixed_overrides = {

440 "tp": constraint.get("fixed_tp_degree"),

441 "pp": constraint.get("fixed_pp_degree"),

442 "cp": constraint.get("fixed_cp_degree"),

443 "ep": constraint.get("fixed_ep_degree"),

444 }

445

446 # Use the *minimum* product to check that at least one valid

447 # combination fits within the device budget. The enumerator

448 # (the strategy enumerator) will filter invalid combos.

449 min_product = dp_min

450 for search_key, dim_label in dim_keys.items():

451 fixed_val = fixed_overrides.get(dim_label)

452 if fixed_val is not None and fixed_val > 0:

453 min_product *= fixed_val

454 else:

455 candidates = search.get(search_key, [1])

456 if not candidates:

457 candidates = [1]

458 min_product *= min(candidates)

459

460 if min_product > total_devices:

461 errors.append(_err(

462 "search_space",

463 f"Minimum product of parallel dimensions ({min_product}) exceeds "

464 f"total available devices ({total_devices})",

465 ))

466

467

468def _check_memory_limit(

469 errors: List[ValidationError],

470 cluster: Dict,

471 constraint: Dict,

472) -> None:

473 """Check that memory_limit_gb is non-negative and does not exceed device memory."""

474 memory_limit = constraint.get("memory_limit_gb", 0.0)

475 if memory_limit < 0:

476 errors.append(_err(

477 "constraint.memory_limit_gb",

478 f"memory_limit_gb must be >= 0, got {memory_limit}",

479 ))

480

481 device_memory = cluster.get("device_memory_gb", 0.0)

482 if memory_limit > 0 and device_memory > 0:

483 if memory_limit > device_memory:

484 errors.append(_warn(

485 "constraint.memory_limit_gb",

486 f"memory_limit_gb ({memory_limit}) exceeds "

487 f"device_memory_gb ({device_memory})",

488 ))

489

490

491def _check_dense_model_ep_cp_warning(

492 errors: List[ValidationError],

493 model: Dict,

494 search: Dict,

495) -> None:

496 """Issue 126 Section 5 rule 9: warn when EP/CP are enabled on Dense models.

497

498 If ``moe_enabled`` is ``False`` and ``ep_degree`` or ``cp_degree``

499 candidates contain values > 1, emit a warning since EP/CP are

500 typically used for MoE and long-sequence scenarios respectively.

501 """

502 moe_enabled = model.get("moe_enabled", False)

503 if moe_enabled:

504 return

505

506 ep_vals = search.get("expert_parallel_degree", [1])

507 if ep_vals and any(v > 1 for v in ep_vals):

508 errors.append(_warn(

509 "search_space.expert_parallel_degree",

510 "Expert Parallelism (ep > 1) is configured but moe_enabled is "

511 "False. EP has no effect on Dense LLMs.",

512 ))

513

514 cp_vals = search.get("context_parallel_degree", [1])

515 if cp_vals and any(v > 1 for v in cp_vals):

516 errors.append(_warn(

517 "search_space.context_parallel_degree",

518 "Context Parallelism (cp > 1) is configured on a Dense LLM. "

519 "Ensure this is intentional for long-sequence scenarios.",

520 ))

521

522

523def _check_fsdp_hsdp_device_product(

524 errors: List[ValidationError],

525 search: Dict,

526 constraint: Dict,

527 cluster: Dict,

528) -> None:

529 """Issue 127 constraint: FSDP/HSDP device product validation.

530

531 Verifies that the sum of shard-degree and replicate-degree

532 (which together form a complete DP decomposition) does not

533 exceed available devices when combined with TP/PP/CP/EP.

534 """

535 total_devices = _total_cards(cluster)

536 if total_devices <= 0:

537 return

538

539 fixed_dp = constraint.get("fixed_dp_degree")

540 dp_shard_vals = search.get("data_parallel_shard_degree", [1])

541 dp_repl_vals = search.get("data_parallel_replicate_degree", [1])

542

543 if fixed_dp is not None and fixed_dp > 0:

544 dp_shard_vals = [fixed_dp]

545 dp_repl_vals = [1]

546

547 fixed_overrides = {

548 "tp": constraint.get("fixed_tp_degree"),

549 "pp": constraint.get("fixed_pp_degree"),

550 "cp": constraint.get("fixed_cp_degree"),

551 "ep": constraint.get("fixed_ep_degree"),

552 }

553

554 dim_keys = {

555 "tensor_parallel_degree": "tp",

556 "pipeline_parallel_degree": "pp",

557 "context_parallel_degree": "cp",

558 "expert_parallel_degree": "ep",

559 }

560

561 has_over_product = True

562 for dp_shard in (dp_shard_vals or [1]):

563 for dp_repl in (dp_repl_vals or [1]):

564 product = dp_shard * dp_repl

565 for search_key, dim_label in dim_keys.items():

566 fixed_val = fixed_overrides.get(dim_label)

567 if fixed_val is not None and fixed_val > 0:

568 product *= fixed_val

569 else:

570 candidates = search.get(search_key, [1]) or [1]

571 product *= min(candidates)

572 if product <= total_devices:

573 has_over_product = False

574 break

575 if not has_over_product:

576 break

577

578 if has_over_product:

579 errors.append(_err(

580 "search_space",

581 f"No FSDP/HSDP decomposition (dp_shard * dp_replicate) "

582 f"fits within total available devices ({total_devices}) "

583 f"when combined with TP/PP/CP/EP dimensions.",

584 ))

Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / config_adapter / _constraint_checker.py: 94%

250 statements