Diff Coverage

Diff: origin/master...HEAD, staged and unstaged changes

Source File Diff Coverage (%) Missing Lines
hyper_parallel/auto_parallel/dense_tuner/__init__.py 100%  
hyper_parallel/auto_parallel/dense_tuner/candidate_generator.py 90.7% 256,258,293,364-368,370,412,415,421,495,498-499,511,514-515,583,607,633
hyper_parallel/auto_parallel/dense_tuner/config.py 98.8% 421,467
hyper_parallel/auto_parallel/dense_tuner/estimator.py 97.2% 72,74,326,390,465
hyper_parallel/auto_parallel/dense_tuner/pipeline_balancer.py 89.0% 108,111-113,149,151,272,327,383-384,391,475,568-570,573,595-596,598
hyper_parallel/auto_parallel/dense_tuner/result.py 100%  
hyper_parallel/auto_parallel/dense_tuner/strategy.py 98.9% 156
hyper_parallel/auto_parallel/dense_tuner/tuner.py 91.3% 90,209-210,240,244,249-251,280-281
hyper_parallel/auto_parallel/dense_tuner/visualizer.py 82.7% 61-63,105,112,132,161-162,164,181,210-211,213,256,282-283,285,316-318,333-334,336,355-356,363,366-383,469-470,472
hyper_parallel/auto_parallel/dense_tuner/candidate_generator.py
252
253
254
255
256
257
258
259
260
261
262
        return False
    if search_space.pp_range and strategy.pp_degree not in search_space.pp_range:
        return False
    if search_space.cp_range and strategy.cp_degree not in search_space.cp_range:
        return False
    if search_space.micro_batch_num_range and strategy.micro_batch_num not in search_space.micro_batch_num_range:
        return False
    return True


def generate_candidates_sapp(
289
290
291
292
293
294
295
296
297
    _patch_evaluator_set_config(parallelize)
    parallelize.instance.enable_debug = False
    ccfg = parallelize.config.ccfg
    if "vp" not in ccfg.__dict__ or ccfg.__dict__.get("vp", 0) < 1:
        ccfg.vp = 1
    parallelize.bound_space()
    space = parallelize.generate_search_space(
        folder="", threads_num=None
    )
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
    )
    if not os.path.isdir(nd_output_dir):
        return collected

    for pattern in ["results.pdf", "debug_*.csv"]:
        for src in sorted(glob.glob(os.path.join(nd_output_dir, pattern))):
            dst = os.path.join(output_dir, os.path.basename(src))
            shutil.copy2(src, dst)
            collected.append(dst)

    return collected


def generate_candidates(
    hardware: HardwareConfig,
408
409
410
411
412
413
414
415
416
417
418
419
                continue
            remaining = n // (tp * pp)
            for cp in cp_values:
                if remaining % cp != 0:
                    continue
                dp = remaining // cp
                if dp < 1:
                    continue
                if dp_values and dp not in dp_values:
                    continue

                mbn_values = _enumerate_mbn_values(gbs, dp, pp, search_space)
417
418
419
420
421
422
423
424
425
                    continue

                mbn_values = _enumerate_mbn_values(gbs, dp, pp, search_space)
                if not mbn_values and not search_space.micro_batch_num_range:
                    mbn_values = [1]

                for mbn in mbn_values:
                    strategy = ParallelStrategy(
                        dp_degree=dp,
491
492
493
494
495
496
497
498
499
500
501
502
503
            filtered.append(s)
            continue

        if s.tp_degree > kv_heads:
            s.mark_infeasible(
                f"tp_degree={s.tp_degree} > num_kv_heads={kv_heads}"
            )
            filtered.append(s)
            continue

        if s.pp_degree > 1 and model.num_layers % s.pp_degree != 0:
            s.mark_infeasible(
                f"num_layers={model.num_layers} not divisible by "
507
508
509
510
511
512
513
514
515
516
517
518
519
            continue

        total = s.dp_degree * s.tp_degree * s.pp_degree * s.cp_degree
        if total != n:
            s.mark_infeasible(
                f"dp*tp*pp*cp={total} != num_devices={n}"
            )
            filtered.append(s)
            continue

        if s.pp_degree > 1 and s.micro_batch_num < s.pp_degree:
            s.mark_infeasible(
                f"micro_batch_num={s.micro_batch_num} < pp_degree={s.pp_degree}"
579
580
581
582
583
584
585
586
587
        return strategies

    for s in strategies:
        if not s.is_feasible:
            continue
        if s.memory_cost.total > constraints.memory_limit_mb:
            s.mark_infeasible(
                f"memory={s.memory_cost.total:.0f}MB > "
                f"limit={constraints.memory_limit_mb:.0f}MB"
603
604
605
606
607
608
609
610
611
        Filtered list of strategies.
    """
    for s in strategies:
        if not s.is_feasible:
            continue
        if s.dp_degree < constraints.min_dp_degree:
            s.mark_infeasible(
                f"dp_degree={s.dp_degree} < min_dp_degree={constraints.min_dp_degree}"
            )
629
630
631
632
633
634
635
636
637
        return strategies

    for s in strategies:
        if not s.is_feasible:
            continue
        if s.performance_cost.total > constraints.max_step_time_ms:
            s.mark_infeasible(
                f"step_time={s.performance_cost.total:.1f}ms > "
                f"max_step_time={constraints.max_step_time_ms:.1f}ms"
hyper_parallel/auto_parallel/dense_tuner/config.py
417
418
419
420
421
422
423
424
425
            extra["dimensions"] = dimensions
        par = Parallelize("hyper_v2", yaml_path, machine, **extra)
        ccfg = par.config.ccfg
        if not hasattr(ccfg, "vp") or ccfg.vp == 0:
            ccfg.vp = 1
        par._yaml_path = yaml_path
        return par

    def to_sapp_yaml(self) -> str:
463
464
465
466
467
468
469
470
471
        mem_gb = int(h.memory_per_device_mb / 1024)

        ac_mode = "full"
        if hasattr(self, "_recompute_mode"):
            ac_mode = self._recompute_mode

        doc = {
            "model": {
                "name": m.model_name,
hyper_parallel/auto_parallel/dense_tuner/estimator.py
68
69
70
71
72
73
74
75
76
77
78
    new_pp = strategy.pp_degree
    new_vpp = strategy.vpp_degree
    n_lay = ccfg.n_lay
    if getattr(ccfg, "emb_out_in_offset", False):
        n_lay += 2
    if getattr(ccfg, "is_mtp_in_offset", False):
        n_lay += getattr(ccfg, "n_mtp", 0)
    adapter = BalancingAdapter(
        layers=n_lay,
        offset=ccfg.offset,
        recompute=ccfg.full_rec,
322
323
324
325
326
327
328
329
        )

        _apply_strategy_to_ccfg(ccfg, strategy)
        device_type = hardware.to_sapp_machine().device
        sapp_score = sapp_estimate_performance(ccfg, device_type=device_type)

    pp = strategy.pp_degree
    mbn = strategy.micro_batch_num
386
387
388
389
390
391
392
393
394
    total_flops = n_lay * flops_per_layer / pp

    tflops = hardware.tflops_per_device
    if tflops <= 0:
        tflops = 300.0
    compute_ms = total_flops / (tflops * 1e9) * 1000

    comm_ms = 0.0
    if tp > 1:
461
462
463
464
465
466
467
468
469
    stages = []
    start = 0
    for i in range(pp):
        if offset and i < len(offset):
            num_layers = layers_per_stage + int(offset[i])
        else:
            num_layers = layers_per_stage
        end = start + num_layers
        stage_mem = strategy.memory_cost.total / pp
hyper_parallel/auto_parallel/dense_tuner/pipeline_balancer.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
        FileNotFoundError: If *profiling_json_path* does not exist.
        ValueError: If the JSON file is missing ``layers_description``.
    """
    if profiling_json_path:
        from hyper_parallel.auto_parallel.sapp_ppb.utils.layer import (
            generate_layers_list,
        )
        folder = os.path.dirname(os.path.abspath(profiling_json_path))
        base = os.path.splitext(os.path.basename(profiling_json_path))[0]
        return generate_layers_list(folder, base)

    from hyper_parallel.auto_parallel.sapp_nd.memory_estimation.estimate_v2 import (
        EvaluatorV2,
    )
145
146
147
148
149
150
151
152
153
154
155
    model_name = getattr(ccfg, "model_name", "dense-llm")

    layer_list = ppb_input.get("layers_description", [])
    if not layer_list:
        layer_list = ppb_input.get("layers_description_new", [])
    if not layer_list:
        return layers

    for layer_dict in layer_list:
        ltype_str = layer_dict.get("type", "BODY")
        ltype = {
268
269
270
271
272
273
274
275
276
    new_stage_info = []
    for i in range(min(n_stages, len(mem_act_flat))):
        n_lay = strategy.extra.get("layers_per_stage", strategy.extra.get("num_layers", 0) // n_stages)
        if isinstance(n_lay, list) and i < len(n_lay):
            nl = n_lay[i]
        else:
            nl = strategy.memory_cost.total / n_stages if n_stages > 0 else 0
        end = start + int(nl) if isinstance(nl, (int, float)) else start
        new_stage_info.append(StageInfo(
323
324
325
326
327
328
329
330
331
    stage_recompute = [0] * num_stages
    for layer_name in body_names:
        rec_map = variables.get(layer_name)
        if rec_map is None:
            continue
        for rec in Recompute.TYPE:
            counts = rec_map[rec]
            for i in range(num_interleave):
                for s in range(num_stages):
379
380
381
382
383
384
385
386
387
388
            number_of_micro_batch=num_micro_batch,
            max_memory=max_memory,
            layers=layers,
        )
    except Exception:
        pipeline = _solve_pipeline(
            model_name=model_name,
            num_stages=num_stages,
            num_micro_batch=num_micro_batch,
            max_memory=max_memory,
387
388
389
390
391
392
393
394
395
            num_micro_batch=num_micro_batch,
            max_memory=max_memory,
            layers=layers,
        )
        best_dist = pipeline.get_result()

    pipeline = _solve_pipeline(
        model_name=model_name,
        num_stages=num_stages,
471
472
473
474
475
476
477
478
479
        strategy, ccfg, profiling_json_path=profiling_json_path,
    )

    if not layers:
        raise RuntimeError(
            f"PPB produced no layers for strategy {strategy.key}"
        )

    max_memory = int(memory_limit_mb) if memory_limit_mb else int(strategy.memory_cost.total)
564
565
566
567
568
569
570
571
572
573
574
575
576
    _apply_strategy_to_ccfg(ccfg, strategy)

    try:
        layers = _build_layers_from_strategy(strategy, ccfg)
    except Exception as exc:
        logger.warning("PPB layer build failed for %s plot: %s", strategy.key, exc)
        return None

    if not layers:
        return None

    max_memory = int(memory_limit_mb) if memory_limit_mb else int(strategy.memory_cost.total)
    vpp = strategy.vpp_degree if strategy.vpp_degree > 0 else 1
591
592
593
594
595
596
597
598
        plot_path = os.path.join(output_dir, f"ppb_pipeline_{strategy.key}.svg")
        pipeline.simulate(show=False, file_name=plot_path)
        if os.path.exists(plot_path):
            return plot_path
    except Exception as exc:
        logger.warning("PPB pipeline plot failed for %s: %s", strategy.key, exc)

    return None
hyper_parallel/auto_parallel/dense_tuner/strategy.py
152
153
154
155
156
157
158
159
160
            f"cp{self.cp_degree}",
            f"mbn{self.micro_batch_num}",
        ]
        if self.vpp_degree > 1:
            parts.append(f"vpp{self.vpp_degree}")
        if self.ep_degree > 1:
            parts.append(f"ep{self.ep_degree}")
        if self.op_degree > 1:
            parts.append(f"op{self.op_degree}")
hyper_parallel/auto_parallel/dense_tuner/tuner.py
86
87
88
89
90
91
92
93
94

    def _init_sapp(self) -> None:
        """Lazily initialise SAPP-ND Parallelize instance and extract ccfg."""
        if self._parallelize is not None:
            return
        self._parallelize = self.config.create_sapp_parallelize()
        self._ccfg = self._parallelize.config.ccfg

    def tune(self) -> TunerResult:
205
206
207
208
209
210
211
212
213
214
                try:
                    nd_plot_files.extend(
                        generate_nd_memory_plots(s, self._ccfg, out)
                    )
                except Exception as exc:
                    logger.warning("ND memory plot failed: %s", exc)

        chart_files = self._visualizer.visualize(
            result.top_k_strategies,
            result.all_candidates,
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
            constraints: Constraint configuration.
        """
        gbs = constraints.global_batch_size
        if gbs <= 0:
            gbs = self._infer_gbs(strategies, hardware)

        for s in strategies:
            if not s.is_feasible:
                continue
            try:
                s.memory_cost = estimate_memory(s, model, gbs, ccfg=self._ccfg)
                s.performance_cost = estimate_performance(s, model, hardware, gbs, ccfg=self._ccfg)
                s.stage_info = compute_stage_info(s, model, gbs)
            except Exception as exc:
                s.mark_not_supported(f"estimation failed: {exc}")
                logger.warning("Estimation failed for %s: %s", s.key, exc)

    def _infer_gbs(
        self,
        strategies: List[ParallelStrategy],
276
277
278
279
280
281
282
283
284
285
        yaml_path = getattr(self._parallelize, "_yaml_path", None)
        if yaml_path and os.path.exists(yaml_path):
            try:
                os.unlink(yaml_path)
            except OSError:
                pass

    @staticmethod
    def _build_ppb_result(strategy: ParallelStrategy) -> Optional[PPBResult]:
        """Extract PPB result from a strategy's extra metadata.
hyper_parallel/auto_parallel/dense_tuner/visualizer.py
57
58
59
60
61
62
63
64
65
66
            try:
                import matplotlib
                matplotlib.use("Agg")
                self._matplotlib_available = True
            except ImportError:
                self._matplotlib_available = False
                logger.warning(
                    "matplotlib not available; falling back to text charts"
                )
        return self._matplotlib_available
101
102
103
104
105
106
107
108
109
        for src in (nd_plot_files or []):
            if src and os.path.exists(src):
                dst = os.path.join(out, os.path.basename(src))
                if os.path.abspath(src) != os.path.abspath(dst):
                    shutil.copy2(src, dst)
                files.append(dst)

        for src in (ppb_plot_files or []):
            if src and os.path.exists(src):
108
109
110
111
112
113
114
115
        for src in (ppb_plot_files or []):
            if src and os.path.exists(src):
                dst = os.path.join(out, os.path.basename(src))
                if os.path.abspath(src) != os.path.abspath(dst):
                    shutil.copy2(src, dst)
                files.append(dst)

        return [f for f in files if f]
128
129
130
131
132
133
134
135
136
        Returns:
            Output file path, or None if no strategies.
        """
        if not strategies:
            return None

        labels = [s.key for s in strategies]
        step_times = [s.performance_cost.total for s in strategies]
        path = os.path.join(out, "step_time_comparison.txt")
157
158
159
160
161
162
163
164
165
166
167
168
                img_path = os.path.join(out, "step_time_comparison.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _memory_chart(
        self,
        strategies: List[ParallelStrategy],
177
178
179
180
181
182
183
184
185
        Returns:
            Output file path, or None if no strategies.
        """
        if not strategies:
            return None

        labels = [s.key for s in strategies]
        memories = [s.memory_cost.total for s in strategies]
        path = os.path.join(out, "memory_comparison.txt")
206
207
208
209
210
211
212
213
214
215
216
217
                img_path = os.path.join(out, "memory_comparison.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _stage_distribution_chart(
        self,
        strategies: List[ParallelStrategy],
252
253
254
255
256
257
258
259
260

                n_strats = len(pp_strategies)
                fig, axes = plt.subplots(1, n_strats, figsize=(6 * n_strats, 4))
                if n_strats == 1:
                    axes = [axes]
                for ax, s in zip(axes, pp_strategies):
                    stages = s.stage_info
                    stage_ids = [st.stage_id for st in stages]
                    mems = [st.memory_mb for st in stages]
278
279
280
281
282
283
284
285
286
287
288
289
                img_path = os.path.join(out, "stage_distribution.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _feasibility_chart(
        self,
        all_candidates: List[ParallelStrategy],
312
313
314
315
316
317
318
319
320
321
322
                f.write(f"Feasible:         {feasible} ({100 * feasible / total:.1f}%)\n")
                f.write(f"Infeasible:       {infeasible} ({100 * infeasible / total:.1f}%)\n")
                f.write(f"Not supported:    {not_supported} ({100 * not_supported / total:.1f}%)\n")
            else:
                f.write("Feasible:         0\n")
                f.write("Infeasible:       0\n")
                f.write("Not supported:    0\n")

        if self._check_matplotlib() and total > 0:
            try:
                import matplotlib.pyplot as plt
329
330
331
332
333
334
335
336
337
338
339
340
                img_path = os.path.join(out, "feasibility_distribution.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _filter_reasons_chart(
        self,
        all_candidates: List[ParallelStrategy],
351
352
353
354
355
356
357
358
359
360
        """
        reasons: Dict[str, int] = {}
        for s in all_candidates:
            if s.filter_reason:
                category = re.split(r"[=<>]", s.filter_reason)[0].strip()
                reasons[category] = reasons.get(category, 0) + 1

        path = os.path.join(out, "filter_reasons.txt")
        with open(path, "w", encoding="utf-8") as f:
            f.write("Filter Reasons Statistics\n")
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
        with open(path, "w", encoding="utf-8") as f:
            f.write("Filter Reasons Statistics\n")
            f.write("=" * 40 + "\n")
            for reason, count in sorted(reasons.items(), key=lambda x: -x[1]):
                f.write(f"  {reason}: {count}\n")

        if self._check_matplotlib() and reasons:
            try:
                import matplotlib.pyplot as plt
                fig, ax = plt.subplots(figsize=(8, max(4, len(reasons) * 0.5)))
                sorted_reasons = sorted(reasons.items(), key=lambda x: -x[1])
                labels = [r[0][:40] for r in sorted_reasons]
                counts = [r[1] for r in sorted_reasons]
                ax.barh(range(len(labels)), counts, align="center")
                ax.set_yticks(range(len(labels)))
                ax.set_yticklabels(labels)
                ax.set_xlabel("Count")
                ax.set_title("Filter Reasons Distribution")
                fig.tight_layout()
                img_path = os.path.join(out, "filter_reasons.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path

    def _dimension_trend_chart(
465
466
467
468
469
470
471
472
                img_path = os.path.join(out, "dimension_trends.png")
                fig.savefig(img_path, dpi=150)
                plt.close(fig)
                return img_path
            except Exception as exc:
                logger.warning("matplotlib chart failed: %s", exc)

        return path