Diff Coverage

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

Source File Diff Coverage (%) Missing Lines
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_backbone.py 79.8% 239,626,689,695-696,714-715,727-728,730-734,739,754-755,758,761
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_bwd_overhead.py 91.3% 132,184
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_context.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_hook_manager.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/estimate_v2.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/body.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/comm.py 91.0% 247,448-449,451-452,454-455
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/head.py 79.2% 68,101-104
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/tail.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/utils.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/balancing_adapter.py 75.0% 42
hyper_parallel/auto_parallel/sapp_nd/nd/common/_cost_model_variables.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/arch_hooks.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/cost_model_preprocess.py 89.3% 286,300-304,385,403
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/_cost_model_parser.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_hyper.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_hyperparallel.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_mindformers.py 85.7% 175,185,189,191,202
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_mindspeed.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/hardware.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/debug.py 85.7% 72,616
hyper_parallel/auto_parallel/sapp_nd/nd/dimensions.py 75.6% 33-35,252,298-299,307-308,310-311,313,320-321,323-324,326-327,341,343-344
hyper_parallel/auto_parallel/sapp_nd/nd/global_config.py 58.8% 274-280
hyper_parallel/auto_parallel/sapp_nd/nd/parallelize.py 75.9% 317,319,321-322,441-442,479-484,487
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/comm_time.py 83.2% 70,110,113,149-154,419-421,435,444,476,481,501,518,542,545,594-597,667,895-896,908-909,941-947,949-951,999,1013,1015-1018,1022,1024,1030
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/estimate.py 91.9% 59,62,67-68,73,76,159-162
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_backbone.py
235
236
237
238
239
240
241
242
243
                real_id = self._ctx.real_lay_ids[chunk_id][i][lay_id]
                name = f"Lay_{real_id}_{lay_type.name[0]}"
                idx = chunk_id * len(dyn_mem_i) + lay_id
                stat = self.mb(stat_mem_i[chunk_id][lay_id])
                dyn = self.mb(sum(dyn_mem_i[chunk_id][lay_id]))
                ax.bar(
                    0.1,
                    [dyn],
                    bottom=bottoms["dyn"],
622
623
624
625
626
627
628
629
630
                    sm["stat"][stage_id][chunk_id][lay_id] = static_mem
                    dyn_val = self._inner_dynamic_mem()
                    sm["dyn"][stage_id][chunk_id][lay_id] = dyn_val
                    if os.environ.get("_SAPP_DEBUG_DYN") == "1":
                        _sys.stderr.write(
                            f"DEBUG_DYN stage={stage_id} chunk={chunk_id} lay={lay_id} "
                            f"node={node} active={dyn_val[0] if isinstance(dyn_val, tuple) else dyn_val} "
                            f"comm={dyn_val[1] if isinstance(dyn_val, tuple) else 0} "
                            f"mf={self._ctx.micro_factor} h={self._ccfg.h} "
685
686
687
688
689
690
691
692
693
        if fw_oh > 0 and d_shard > 1:
            if d_replicate > 1:
                fw_oh = int(fw_oh * (1.0 + hsdp_coeff * (d_shard - 1)))
            else:
                fw_oh = int(fw_oh * (1.0 + fsdp_coeff * (d_shard - 1)))
        if fw_oh > 0 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + rep_discount * (d_replicate - 1)))
        if fw_oh > 0 and d_shard > 1 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + shard_rep_interact * (d_shard - 1) * (d_replicate - 1) / d_shard))
691
692
693
694
695
696
697
698
699
700
            fw_oh = int(fw_oh / (1.0 + rep_discount * (d_replicate - 1)))
        if fw_oh > 0 and d_shard > 1 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + shard_rep_interact * (d_shard - 1) * (d_replicate - 1) / d_shard))
        if fw_oh > 0 and d_replicate > 1 and tp >= tp_threshold:
            tp_hsdp_extra = int(fw_oh * tp_extra_ratio)
            fw_oh += tp_hsdp_extra
        return fw_oh

    def __build_stage_insight(
        self, stage_id, raw_dyn, total_raw, total_fw_oh,
710
711
712
713
714
715
716
717
718
719
            fw_oh_share = total_fw_oh * (
                0.5 / num_stages + 0.5 * raw_dyn[stage_id] / total_raw
            )
            ins["Dynamic"] += int(fw_oh_share)
        elif ins["Dynamic"] > 0:
            ins["Dynamic"] += total_fw_oh // num_stages
        stage_accu = sm["logs"][stage_id].accu_mem_type
        ins["ModelParameters"] = self.mb(stage_accu[MemType.MODEL_PARAM])
        ins["OptimizerStates"] = self.mb(stage_accu[MemType.OPTIM_STATE])
        ins["AccumulGradients"] = self.mb(stage_accu[MemType.ACCU_GRAD])
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
        ins["AllGather Comm"] = self.mb(stage_accu[MemType.AG_COMM])
        ins["All2All Comm"] = self.mb(stage_accu[MemType.A2A_COMM])

        if self._ccfg.cp > 1:
            cp_memory = EvalBody.act_cp_layer(self._ccfg, self._ctx)
            cp_comm_buffer = EvalLayerComm.cp_comm_buffer(self._ccfg, self._ctx)

            ins["CP KV Cache"] = self.mb(cp_memory.kv_cache_memory)
            ins["CP Attn Scores"] = self.mb(cp_memory.attention_scores_memory)
            ins["CP Softmax"] = self.mb(cp_memory.softmax_outputs_memory)
            ins["CP Comm Buffer"] = self.mb(cp_comm_buffer)
            ins["CP Reduction"] = self.mb(cp_memory.total_reduction)

        ins["Node Log"] = sm["logs"][stage_id].node_compute_log
        # VERBOSE
        if verbose and spec_stage_id in (-1, stage_id):
            self.__verbose_insights(sm, stage_id, ins)
        ins["Static"] = self.mb(ins["Static"])
        ins["Dynamic"] = self.mb(ins["Dynamic"])
        insights.append(ins)
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
        num_stages = self._ccfg.p
        safety_buffer = int(getattr(self._ccfg, 'safety_buffer_gb', 1)) * 1024 * 1024 * 1024
        raw_dyn = self.__compute_raw_dynamic(num_stages, stages, record_lay_types, sm, safety_buffer)
        if os.environ.get("_SAPP_DEBUG_DYN") == "1":
            for sid in range(num_stages):
                act_s = sum(
                    sum(aval for aval, _ in cell) for cell in sm["dyn"][sid]
                )
                comm_p = max(
                    max(comm for _, comm in cell) for cell in sm["dyn"][sid]
                )
                _sys.stderr.write(
                    f"DEBUG_POST stage={sid} act_sum={act_s/(1024**2):.1f}MB "
                    f"comm_peak={comm_p/(1024**2):.1f}MB "
                    f"raw_dyn={raw_dyn[sid]/(1024**2):.1f}MB\n"
                )
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_bwd_overhead.py
128
129
130
131
132
133
134
135
136
        current_node = getattr(self._ctx, "current_node", None)
        if current_node == LayerType.OUTPUT_LAYER:
            t = getattr(self._ccfg, "t", 1)
            if t > 1:
                bwd_workspace = (
                    (1.0 - 1.0 / t)
                    * getattr(self._ccfg, "s", 0)
                    * getattr(self._ccfg, "b", 0)
                    * getattr(self._ccfg, "v", 0)
180
181
182
183
184
185
186
187
                        self._ctx.current_lay_id = (
                            f"G_{self._ctx.current_lay_id}"
                        )
                        self._ctx.current_node = last_mtp
                        res += self.__bwd_dyn_mem(
                            is_recomputed=False, default_micro_factor=1
                        )
        return res
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/comm.py
243
244
245
246
247
248
249
250
251
        Result is in bytes (like TP activation comm), unlike CP/DP which are
        in element counts (parameter comm).
        """
        if ccfg.comm_ep == 0:
            return 0
        if ccfg.n_exp <= 1:
            return 0
        if ccfg.ep <= 1:
            return 0
444
445
446
447
448
449
450
451
452
453
454
455
    def tp_buffer_comm(ccfg: CostModelConfig, ctx: Context, mb: int) -> float:
        """Peak HBM buffer for TP all-gather (bytes)."""
        if ccfg.t <= 1 or ccfg.comm_t == 0:
            return 0
        rec_layer = ctx.current_node == LayerType.SEL_REC_LAYER
        rec_coeff = EvalUtils.rec_coeff(rec_layer, ccfg.rec_op.gather)

        attn_buf = ccfg.s * ccfg.b * ccfg.h * ccfg.bytes_compute
        ffn_buf = ccfg.s * ccfg.b * 2 * ccfg.hff * ccfg.bytes_compute

        peak_buf = max(attn_buf, ffn_buf)
        return rec_coeff * ccfg.comm_t * mb * peak_buf / ccfg.cp
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/head.py
64
65
66
67
68
69
70
71
72

    @staticmethod
    def dp_comm_embed(ccfg: CostModelConfig, ctx: Context) -> float:
        if getattr(ccfg, "fsdp", False):
            return 0.0
        return (
            ccfg.comm_d_non_exp
            * ctx.eval.num_p(ccfg, ctx)
            / (ccfg.shard_embed * ccfg.cp)
 97
 98
 99
100
101
102
103
104
105
106
107
108
    @staticmethod
    def hsdp_comm_embed(ccfg: CostModelConfig, ctx: Context) -> float:
        if getattr(ccfg, "comm_hsdp", 0) <= 0:
            return 0.0
        param_size = ctx.eval.num_p(ccfg, ctx)
        d_shard = ccfg.d_shard_or_d
        sharded_size = param_size / (d_shard * ccfg.cp * ccfg.t)
        return ccfg.comm_hsdp * sharded_size * ccfg.bytes_compute

    @staticmethod
    def fsdp_grad_comm_embed(ccfg: CostModelConfig, ctx: Context) -> float:
        param_size = ctx.eval.num_p(ccfg, ctx)
hyper_parallel/auto_parallel/sapp_nd/nd/balancing_adapter.py
38
39
40
41
42
43
44
45
46
        return self.pp * self.vpp

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Pipeline):
            return NotImplemented
        return self.pp == other.pp and self.vpp == other.vpp


def infer_pp_and_vpp(offset):
hyper_parallel/auto_parallel/sapp_nd/nd/common/cost_model_preprocess.py
282
283
284
285
286
287
288
289
290
            self.__maybe_set_int(target_ccfg, attr, value)
        if target_ccfg.use_seq_parallel:
            target_ccfg.sp = target_ccfg.t if target_ccfg.t > 1 else 1
        else:
            target_ccfg.sp = 1
        if target_ccfg.recompute_slice_activation and tp is not None:
            target_ccfg.shard_recompute_input = target_ccfg.t
            target_ccfg.shard_output_activ = target_ccfg.t
        if op and isinstance(op, int):
296
297
298
299
300
301
302
303
304
305
306
307
308
        if fsdp_kw is True:
            target_ccfg.fsdp = True
            target_ccfg.has_op = True
        elif fsdp_kw is False:
            target_ccfg.fsdp = False
            if op and isinstance(op, int) and op <= 1:
                target_ccfg.has_op = False
            elif not op:
                target_ccfg.has_op = False

    def __apply_d_shard_strategy(self, target_ccfg, d_shard_kw):
        if d_shard_kw is not None and isinstance(d_shard_kw, int) and d_shard_kw > 1:
            target_ccfg.d_shard = d_shard_kw
381
382
383
384
385
386
387
388
389
            and isinstance(d_shard_kw, int)
            and 1 < d_shard_kw < target_ccfg.d
        )
        if is_hsdp:
            target_ccfg.fsdp = True
        saved_d_shard = target_ccfg.d_shard

        logger.debug(
            "in ccfg: DP = %d, TP(MP) = %d, EP = %d, CP = %d, "
399
400
401
402
403
404
405
406
407
            target_ccfg.fsdp,
        )
        self.__apply_parser_config(target_ccfg)
        if is_hsdp:
            target_ccfg.d_shard = saved_d_shard
        self.__apply_recompute_offset(target_ccfg, off, fr, sr)
        self.__maybe_set_int(target_ccfg, "cp", cp)

    def get_strategy(self):
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_mindformers.py
171
172
173
174
175
176
177
178
179
        self.ccfg.bytes_p = self.ccfg.fp_bytes(
            self.config.model.model_config.param_init_type,
        )  # parameters
        if not self.ccfg.bytes_p:
            self.ccfg.bytes_p = self.ccfg.fp_bytes(
                self.config.model.model_config.params_dtype
            )
        self.ccfg.bytes_compute = self.ccfg.fp_bytes(
            self.config.model.model_config.compute_dtype
181
182
183
184
185
186
187
188
189
190
191
192
193
194
        self.ccfg.bytes_softmax = self.ccfg.fp_bytes(
            self.config.model.model_config.softmax_compute_type
        )  # softmax output
        if not self.ccfg.bytes_softmax:
            self.ccfg.bytes_softmax = self.ccfg.fp_bytes(
                self.config.model.model_config.softmax_compute_dtype
            )
        if not self.ccfg.bytes_p:
            raise AttributeError("bytes_p not positive")
        if not self.ccfg.bytes_compute:
            raise AttributeError("bytes_compute not positive")
        self.ccfg.bytes_grad = 4
        self.ccfg.bytes_os = 4
        self.ccfg.bytes_norm = 4
198
199
200
201
202
203
204
205
206
            self.ccfg.framework_overhead = int(
                10.5 * 1024 ** 3 * (cap_gb / 32.0) ** 1.74
            )
        else:
            self.ccfg.framework_overhead = getattr(
                self.ccfg, "framework_overhead", 0
            )

    def __config_parse_yaml_optimizer_shard(self):
hyper_parallel/auto_parallel/sapp_nd/nd/debug.py
68
69
70
71
72
73
74
75
76
            name = "EP"
        elif self == self.CP_COMM:
            name = "CP"
        elif self == self.FSDP_COMM:
            name = "FSDP"
        elif self == self.PP_COMM:
            name = "P2P"
        elif self == self.BUBBLE:
            name = "BBL"
612
613
614
615
616
617
618
619
620
        parts[RealParts.DP_WAIT][-1] += real["op_wait"]

    tp = "tp_wait"
    if tp in real.keys():
        parts[RealParts.MP_WAIT][-1] += real["tp_wait"]

    sp = "sp_wait"
    if sp in real.keys():
        parts[RealParts.MP_WAIT][-1] += real["sp_wait"]
hyper_parallel/auto_parallel/sapp_nd/nd/dimensions.py
29
30
31
32
33
34
35
36
37
38
39
    from hyper_parallel.auto_parallel.sapp_nd.nd.common._cost_model_variables import _CostModVar


def from_str_bool(value: float) -> bool:
    if isinstance(value, str):
        return value.strip().lower() not in ("false", "0", "no", "off", "")
    return bool(value)


class Dimension:
    """Output dimension"""
248
249
250
251
252
253
254
255
256
            valid = dims_val[MBN] >= dims_val[PP]
            valid = valid and not (dims_val[PP] == 1 and dims_val[MBN] > 1)
            if not valid:
                return False
        return True

    @staticmethod
    def _check_power_of_two(dim, value):
        if not (value & (value - 1)) == 0:
294
295
296
297
298
299
300
301
302
303
    def _validate_op_power_of_two(self):
        if OP in self.all_dims:
            op = self.dims_val[OP]
            if not (op & (op - 1)) == 0:
                logger.warning("OP %d must be a power of 2", op)
                return False
        return True

    def _validate_fsdp_constraints(self):
        """Validate FSDP coexistence with OP, EP and the DP lower bound."""
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
        """Validate FSDP coexistence with OP, EP and the DP lower bound."""
        if not (self.has_dim(FSDP) and self.dims_val[FSDP]):
            return True
        if self.has_dim(OP) and self.dims_val[OP] > 1:
            logger.warning("FSDP and OP cannot coexist (OP > 1)")
            return False
        if self.has_dim(EP) and self.dims_val[EP] > 1:
            logger.warning("FSDP and EP cannot coexist")
            return False
        if self.has_dim(DP) and self.dims_val[DP] < 2:
            logger.warning("FSDP requires DP >= 2")
            return False
        return True

    def _validate_hsdp_fsdp_coexistence(self, d_shard):
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331

    def _validate_hsdp_fsdp_coexistence(self, d_shard):
        """Validate HSDP constraints that depend on FSDP being present."""
        if d_shard > 1 and not self.dims_val[FSDP]:
            logger.warning("HSDP d_shard > 1 requires FSDP enabled")
            return False
        if d_shard > 1 and self.dims_val[EP] > 1:
            logger.warning("HSDP and EP cannot coexist")
            return False
        if d_shard > 1 and self.dims_val[OP] > 1:
            logger.warning("HSDP and OP cannot coexist (OP > 1)")
            return False
        return True

    def _validate_hsdp_constraints(self):
        """Validate HSDP-related dimension constraints."""
337
338
339
340
341
342
343
344
345
346
347
348
            return False
        if self.has_dim(FSDP) and not self._validate_hsdp_fsdp_coexistence(
            d_shard
        ):
            return False
        if d_shard > 1 and not (d_shard & (d_shard - 1)) == 0:
            logger.warning("HSDP d_shard %d must be a power of 2", d_shard)
            return False
        return True

    def is_valid(self):
        """Check if all dimensions values are valid"""
hyper_parallel/auto_parallel/sapp_nd/nd/global_config.py
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
            return [False, True]
        return [dim.from_config(self.ccfg)]

    def hsdp_space(self, dp: int) -> List[Any]:
        if Dim.HSDP not in self.dimensions:
            return [dp]
        power_of_2_divisors = Hard.all_divisors(dp)
        valid = [s for s in power_of_2_divisors if 2 <= s < dp]
        if not valid:
            return [dp]
        return [dp] + valid

    def max_op(self, dp, tp, ep):
        """Compute bound for dimension OP"""
        if (
hyper_parallel/auto_parallel/sapp_nd/nd/parallelize.py
313
314
315
316
317
318
319
320
321
322
323
324
325
326
    def _parallel_loops_fsdop(self, space, pool, dtpc_p, mbsn, ep, vpp, dp, tp):
        """Iterate over FSDP/OP/d_shard/SP sub-space for one (ep, vpp)."""
        for fsdp in self.config.bool_space(Dim.FSDP):
            if fsdp and dp < 2:
                continue
            if fsdp and ep > 1:
                continue
            if fsdp:
                op_space = [1]
                d_shard_space = self.config.hsdp_space(dp)
            else:
                op_space = self.config.space(
                    Dim.OP, self.config.max_op(dp, tp, ep)
                )
437
438
439
440
441
442
443
444
445
446
                            memory=mem,
                            cache_file=cache_file,
                        )
                        debugger.write()
                        debugger.info.pop(Debug.PerfParts.TOTAL, None)
                        debugger.info.pop(Debug.PerfParts.MEMORY, None)
                        debug_parts = list(debugger.info.keys())
                        values = list(debugger.info.values())
                    else:
                        score = estimate_performance(
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
        # If MB (microbatch number) is not in the config dimensions,
        # inject it so set_strategy can compute m from gbs/d/b.
        injected_mbn = False
        if Dim.MBN not in config.dims_val and Dim.MBS in config.dims_val and Dim.DP in config.dims_val:
            dp_val = config.dims_val[Dim.DP]
            mbs_val = config.dims_val[Dim.MBS]
            if isinstance(dp_val, int) and isinstance(mbs_val, int) and dp_val > 0 and mbs_val > 0:
                mbn = self.global_batch_size // dp_val // mbs_val
                config.dims_val[Dim.MBN] = mbn
                injected_mbn = True
        self.config.set_parallel_config(config)
        if injected_mbn:
            del config.dims_val[Dim.MBN]
        peak_mem = self.memory_estim()
        est_kwargs = {
            "debugger": debugger,
            "device_type": self.machine.device,
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/comm_time.py
66
67
68
69
70
71
72
73
74
        dim_coeffs = fallback.flop_coeffs.get(dimension, {})
        result = dim_coeffs.get(sub_key, {})
        if result:
            return result
    return {}


def _compute_hsdp_features(cfg, d_shard_val, device_type, mb=1):
    """Compute shared derived features for HSDP/FSDP FLOP-mode models."""
106
107
108
109
110
111
112
113
114
115
116
117
def _flop_mode_comp_comm(cfg, d_shard_val, device_type=None, mb=1):
    """Estimate COMP cost for FLOP mode when d_replicate > 1 (HSDP)."""
    f = _compute_hsdp_features(cfg, d_shard_val, device_type, mb=mb)
    if f["d_replicate"] <= 1:
        return 0.0
    c = _get_flop_coeffs(device_type, "comp", "hsdp")
    if not c:
        return 0.0
    total = (
        c.get("INTERCEPT", 0)
        + c.get("TP", 0) * f["tp"]
        + c.get("INV_TP", 0) * f["inv_tp"]
145
146
147
148
149
150
151
152
153
154
155
156
157
158

    Returns:
        Effective bandwidth in bytes/s.
    """
    msg_mb = msg_bytes / 1e6
    if msg_mb <= 0:
        return peak_bw_gbps * 1e9 * small_eff
    t = max(0.0, min(1.0, (log10(max(msg_mb, 1e-6)) - _LOG10_MB_FLOOR) / _LOG10_MB_RANGE))
    eff = small_eff + (large_eff - small_eff) * t
    return peak_bw_gbps * 1e9 * eff


def _cp_resolve_topology(cp, device_per_node, bw_intra, bw_inter):
    """Resolve CP topology and effective bandwidth.
415
416
417
418
419
420
421
422
423
424
425
        if cfg.n_exp > 1:
            fsdp_intra_vol += exp * bc * 2
        d_shard_local = cfg.d_shard_or_d
        if getattr(cfg, "comm_hsdp", 0) > 0 and d_shard_local < cfg.d:
            sharded_non_exp = non_exp / (d_shard_local * cfg.cp * cfg.t)
            sharded_exp = exp / (d_shard_local * cfg.cp * cfg.t_exp) if cfg.n_exp > 1 else 0
            hsdp_inter_vol += (sharded_non_exp + sharded_exp) * bc
    fsdp_layer_count = 1 if is_fsdp_layer else 0

    comm[Dim.TP] += EvalLayerComm.tp_comm_layer(cfg, ctx, 1)
    tp_layer_count += 1
431
432
433
434
435
436
437
438
439

def _flop_mode_fsdp_comm(cfg, fsdp_layer_count, d_shard_val, device_type, pp=1, mb=1):
    """Estimate FSDP/HSDP communication cost for FLOP mode."""
    if fsdp_layer_count <= 0 or d_shard_val <= 0:
        return 0.0

    f = _compute_hsdp_features(cfg, d_shard_val, device_type, mb=mb)
    f["pp"] = max(pp, 1)
    is_hsdp = f["d_replicate"] > 1
440
441
442
443
444
445
446
447
448

    sub_key = "hsdp" if is_hsdp else "fsdp"
    c = _get_flop_coeffs(device_type, "shard", sub_key)
    if not c:
        return 0.0

    if is_hsdp:
        total = (
            c.get("INTERCEPT", 0)
472
473
474
475
476
477
478
479
480
481
482
483
484
485
    f = _compute_hsdp_features(cfg, d_shard_val, device_type, mb=mb)

    if (cfg.fsdp or d_shard_val > 1) and d_shard_val > 0:
        if f["sg"] <= 1:
            return 0.0

        if f["d_replicate"] > 1:
            c = _get_flop_coeffs(device_type, "dp", "hsdp")
            if not c:
                return 0.0
            dp_val = f["d_replicate"] * f["d_shard"]
            total = (
                c.get("INTERCEPT", 0)
                + c.get("TP", 0) * f["tp"]
497
498
499
500
501
502
503
504
505
                f["cross"], score, f["m"], comm_dp_raw,
            )
            return score

        return 0.0

    logger.info(
        "FLOP_DP_FALLBACK: d=%d tp=%d fsdp=%s d_shard=%d raw=%.4f result=0.0",
        f["d"], f["tp"], cfg.fsdp, f["d_shard"],
514
515
516
517
518
519
520
521
522
    if f["d_replicate"] <= 1:
        return 0.0
    c = _get_flop_coeffs(device_type, "tp", "hsdp")
    if not c:
        return 0.0
    total = (
        c.get("INTERCEPT", 0)
        + c.get("INV_TP", 0) * f["inv_tp"]
        + c.get("INV_SG", 0) * f["inv_sg"]
538
539
540
541
542
543
544
545
546
547
548
549
def _flop_mode_pp_total_comm(cfg, d_shard_val, device_type, mb=1):
    """Estimate PP_TOTAL cost for FLOP mode when d_replicate > 1 (HSDP)."""
    f = _compute_hsdp_features(cfg, d_shard_val, device_type, mb=mb)
    if f["d_replicate"] <= 1:
        return 0.0
    c = _get_flop_coeffs(device_type, "pp", "hsdp")
    if not c:
        return 0.0
    total = (
        c.get("INTERCEPT", 0)
        + c.get("AG_VOL_D_REP", 0) * f["ag_vol_d_rep"]
        + c.get("CROSS_SG", 0) * f["cross_sg"]
590
591
592
593
594
595
596
597
598
599
600
        comm[Dim.TP] = _flop_mode_tp_comm(
            cfg, d_shard_val, param["device_type"], mb=mb,
        )
    else:
        c = _get_flop_coeffs(param["device_type"], "tp", "fsdp")
        if c:
            tp_scaling = (c.get("A", 0) + c.get("B", 0) * f["tp"] + c.get("C", 0) / f["tp"]) / max(f["d"], 1)
            comm[Dim.TP] *= max(1, f["tp"] // dev_per_node) * tp_scaling

    comm[Dim.EP] *= max(1, param["cfg"].ep // dev_per_node)
    comm[Dim.CP] *= max(1, param["cfg"].cp // dev_per_node)
663
664
665
666
667
668
669
670
671
    logger.info(lccfgs)
    param["layer_count"] = 0
    param["idx_lccfg"] = 0
    # ignores comm recomp, to improve
    comms = {Dim.DP: [], Dim.TP: [], Dim.EP: []}
    for stage in param["stages"]:
        comm = {Dim.DP: 0.0, Dim.TP: 0.0, Dim.EP: 0.0}
        for chunk in stage:
            for layer in chunk:
891
892
893
894
895
896
897
898
899
        idx = level.value - 1 if isinstance(level, NetworkLevel) else level - 1
        p2p_eff = getattr(device, 'p2p_efficiency', None)
        if p2p_eff is not None and 0 <= idx < len(p2p_eff):
            return p2p_eff[idx]
        if 0 <= idx < len(device.level_efficiency):
            return device.level_efficiency[idx]
    raise ValueError(
        f"No efficiency for level {level}; device required"
    )
904
905
906
907
908
909
910
911
912
        idx = level.value - 1 if isinstance(level, NetworkLevel) else level - 1
        p2p = getattr(device, 'p2p_bandwidth', None)
        if p2p is not None and idx < len(p2p):
            return p2p[idx]
        if 0 <= idx < len(device.level_bandwidth):
            return device.level_bandwidth[idx]
    raise ValueError(
        f"No P2P bandwidth for level {level}; device required"
    )
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955


def _shard_group_levels(device, shard_size):
    """Compute message-size-dependent effective bandwidth."""
    remaining = shard_size
    levels = []
    for level in range(device.levels):
        bound = device.level_bound_number[level]
        if bound:
            n = min(remaining, bound)
            remaining = remaining // n if n > 0 else remaining
        else:
            n = remaining
        levels.append(n)
    return levels


def estimate_comm_score(
    cfg,
 995
 996
 997
 998
 999
1000
1001
1002
1003

    d_shard = cfg.d_shard_or_d

    if shard_group_size > 0:
        n_levels = _shard_group_levels(device, shard_group_size)
    else:
        assignment = device.level_assign(
            dp=cfg.d, tp=cfg.t, cp=cfg.cp, pp=cfg.p, d_shard=d_shard
        )
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
        n_level = n_levels[level]
        if n_level <= 1:
            continue
        if per_rank_msg:
            msg_per_rank = comm_volume
        elif shard_group_size > 0:
            prod_from_l = 1
            for k in range(level, len(n_levels)):
                prod_from_l *= n_levels[k]
            msg_per_rank = comm_volume / prod_from_l
        else:
            msg_per_rank = comm_volume / n_level
        if isinstance(a2a_efficiency, dict):
            eff = a2a_efficiency.get(level, level_efficiency(NetworkLevel(level + 1), device=device))
        elif a2a_efficiency is not None:
            eff = a2a_efficiency
        else:
            eff = level_efficiency(NetworkLevel(level + 1), device=device)
        bw_bps = device.level_bandwidth[level] * 1e9 * eff
        lat = level_latency(NetworkLevel(level + 1), device=device)
1026
1027
1028
1029
1030
1031
1032
1033
1034
            eff = level_efficiency(NetworkLevel(level + 1), device=device)
        bw_bps = device.level_bandwidth[level] * 1e9 * eff
        lat = level_latency(NetworkLevel(level + 1), device=device)
        if per_rank_msg:
            collective_time = (
                (n_level - 1) * msg_per_rank / bw_bps
                + lat * 2 * (n_level - 1)
            )
        else:
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/estimate.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80


def efficiency(x):
    """obtained via extrapolation"""
    eff = min(
        1.0, max(0.1, 0.00004694 * x**3 + 0.0014 * x**2 - 0.0336 * x + 0.1)
    )
    return eff


def throughput(precision_bytes, flop):
    """assumes matrix"""
    eff = efficiency(flop / (10.0**12))
    return precision_bytes**2 * (10.0**12) * eff


def estimate_comp_flop_time(cfg, flop, is_softmax=False):
    """flop from throughput"""
    th = throughput(
        cfg.bytes_softmax if is_softmax else cfg.bytes_compute, flop
    )
    return flop / th


def op_table(cfg):
    """OP compute load formulas."""
155
156
157
158
159
160
161
162
163
164
165
166
        if d_replicate > 1:
            mb = cfg.m if hasattr(cfg, 'm') and cfg.m > 0 else 1
            comp_score = _flop_mode_comp_comm(cfg, d_shard_val, device_type=device_type, mb=mb)
            return [comp_score for _ in flops]
        comp_coeffs = _get_flop_coeffs(device_type, "comp", "fsdp")
        comp_tp_scale = (comp_coeffs["A"] + comp_coeffs["B"] / tp_val) * tp_val / d_val
        return [f * comp_tp_scale for f in flops]
    return flops


def estimate_op_bulk_comp(
    cfg,