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 81.0% 230,613,668,674-675,693-694,706-707,709-713,717,732-733,736,739
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_bwd_overhead.py 92.3% 142,192
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_context.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_func_tracer.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_hook_manager.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_ppb.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_utils.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.3% 295,567-568,570-572,574,576-577
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/head.py 85.2% 125-128
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/evaluators/layer_block.py 100%  
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/memory_estimation/hook_base.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/logger.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/score.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/size.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/balancing_adapter.py 96.2% 46
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/config.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/cost_model_preprocess.py 91.0% 282,297-301,389,407
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 86.8% 178,188,192,194,205
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_mindspeed.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/generate_partitions.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/common/hardware.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/debug.py 95.7% 74,612
hyper_parallel/auto_parallel/sapp_nd/nd/dimensions.py 80.8% 45-47,275,335-336,344-345,347-348,350-351,357-358,360-361,363-364,378,380
hyper_parallel/auto_parallel/sapp_nd/nd/global_config.py 77.4% 286-292
hyper_parallel/auto_parallel/sapp_nd/nd/logger.py 100%  
hyper_parallel/auto_parallel/sapp_nd/nd/parallelize.py 79.1% 320,322,324-325,431,436,443-447,479-484,487
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/comm_time.py 83.3% 79,142,145,181-186,451-453,489,498,542,547,567,594,627,630,702-705,821-822,846-847,888-894,896-898,946,960,962-965,969,971,977
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/estimate.py 100%  
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/getters.py 100%  
hyper_parallel/auto_parallel/sapp_nd/perf_estimation/utils_classes.py 100%  
hyper_parallel/auto_parallel/sapp_nd/memory_estimation/_backbone.py
226
227
228
229
230
231
232
233
234
                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"],
609
610
611
612
613
614
615
616
617
                    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} "
664
665
666
667
668
669
670
671
672
        if fw_oh > 0 and d_shard > 1:
            if d_replicate > 1:
                fw_oh = int(fw_oh * (1.0 + 0.03 * (d_shard - 1)))
            else:
                fw_oh = int(fw_oh * (1.0 + 0.01 * (d_shard - 1)))
        if fw_oh > 0 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + 0.025 * (d_replicate - 1)))
        if fw_oh > 0 and d_shard > 1 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + 0.03 * (d_shard - 1) * (d_replicate - 1) / d_shard))
670
671
672
673
674
675
676
677
678
679
            fw_oh = int(fw_oh / (1.0 + 0.025 * (d_replicate - 1)))
        if fw_oh > 0 and d_shard > 1 and d_replicate > 1:
            fw_oh = int(fw_oh / (1.0 + 0.03 * (d_shard - 1) * (d_replicate - 1) / d_shard))
        if fw_oh > 0 and d_replicate > 1 and tp >= 8:
            tp_hsdp_extra = int(fw_oh * 0.28)
            fw_oh += tp_hsdp_extra
        return fw_oh

    def __build_stage_insight(
        self, stage_id, raw_dyn, total_raw, total_fw_oh,
689
690
691
692
693
694
695
696
697
698
            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])
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
        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
        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)
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
        num_stages = self._ccfg.p
        safety_buffer = 1024 * 1024 * 1024  # 1 GB
        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
138
139
140
141
142
143
144
145
146
        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)
188
189
190
191
192
193
194
195
                        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
291
292
293
294
295
296
297
298
299
        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
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
            Peak TP all-gather buffer size in 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
        if ccfg.n_exp > 1:
            ffn_buf = ccfg.s * ccfg.b * 2 * ccfg.hff * ccfg.bytes_compute
        else:
            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
121
122
123
124
125
126
127
128
129
130
131
132
        embedding gradient fragments.
        """
        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:
        """FSDP/HSDP gradient reduce-scatter buffer for embedding params (bytes).
hyper_parallel/auto_parallel/sapp_nd/nd/balancing_adapter.py
42
43
44
45
46
47
48
49
50

    def __eq__(self, other: object) -> bool:
        """Check equality based on pp and vpp values."""
        if not isinstance(other, Pipeline):
            return NotImplemented
        return self.pp == other.pp and self.vpp == other.vpp


def infer_pp_and_vpp(offset: Any) -> Any:
hyper_parallel/auto_parallel/sapp_nd/nd/common/cost_model_preprocess.py
278
279
280
281
282
283
284
285
286
            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):
293
294
295
296
297
298
299
300
301
302
303
304
        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):
        """Apply the HSDP d_shard value, defaulting from dp/fsdp.
385
386
387
388
389
390
391
392
393
            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, "
403
404
405
406
407
408
409
410
411
            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) -> Dict[str, Any]:
hyper_parallel/auto_parallel/sapp_nd/nd/common/framework_parsers/cost_model_parser_mindformers.py
174
175
176
177
178
179
180
181
182
        self.ccfg.bytes_p = self.ccfg.fp_bytes(
            self.config.model.model_config.param_init_type,
        )
        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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
        self.ccfg.bytes_softmax = self.ccfg.fp_bytes(
            self.config.model.model_config.softmax_compute_type
        )
        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
201
202
203
204
205
206
207
208
209
            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
70
71
72
73
74
75
76
77
78
            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"
608
609
610
611
612
613
614
615
616
        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
41
42
43
44
45
46
47
48
49
50
51

    Returns:
        bool: The parsed boolean.
    """
    if isinstance(value, str):
        return value.strip().lower() not in ("false", "0", "no", "off", "")
    return bool(value)


class Dimension:
    """Output dimension"""
271
272
273
274
275
276
277
278
279
            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):
        """Check whether *value* is a power of 2 for the given dimension.
331
332
333
334
335
336
337
338
339
340
        """Validate that OP is a power of 2."""
        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."""
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
        """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):
        """Validate HSDP constraints that depend on FSDP being present."""
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368

    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."""
374
375
376
377
378
379
380
381
382
383
384
            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) -> bool:
hyper_parallel/auto_parallel/sapp_nd/nd/global_config.py
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
        d_shard=1 is excluded because it means "no sharding" which
        contradicts fsdp=True.  When this path is not taken (fsdp=False
        or HSDP dimension absent), d_shard defaults to 1 elsewhere.
        """
        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: int, tp: int, ep: int) -> int:
        """Compute bound for dimension OP"""
        if (
hyper_parallel/auto_parallel/sapp_nd/nd/parallelize.py
316
317
318
319
320
321
322
323
324
325
326
327
328
329
    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)
                )
427
428
429
430
431
432
433
434
435
436
437
438
439
440
                        ),
                    )
                else:
                    if self.enable_debug:
                        debugger = Debug.Debug(
                            config,
                            info_type=Debug.PerfParts,
                            enable=self.enable_debug,
                        )
                        score = estimate_performance(
                            self.config.ccfg,
                            debugger=debugger,
                            device_type=self.machine.device,
                            memory=mem,
439
440
441
442
443
444
445
446
447
448
449
450
451
                            device_type=self.machine.device,
                            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(
                            self.config.ccfg,
                            device_type=self.machine.device,
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
75
76
77
78
79
80
81
82
83
        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.
138
139
140
141
142
143
144
145
146
147
148
149
    Pipeline * m cancels /mb, yielding total. Rank=rank(total).
    """
    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"]
177
178
179
180
181
182
183
184
185
186
187
188
189
190

    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.
447
448
449
450
451
452
453
454
455
456
457
        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
485
486
487
488
489
490
491
492
493
            ag_vol = 1-1/d_shard, dp = d_rep*d_shard.
      rank=10/10, Spearman=1.0, MAPE=0.49%, 10 configs.
    """
    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
494
495
496
497
498
499
500
501
502

    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)
538
539
540
541
542
543
544
545
546
547
548
549
550
551
    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"]
563
564
565
566
567
568
569
570
571
                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"],
590
591
592
593
594
595
596
597
598
    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"]
623
624
625
626
627
628
629
630
631
632
633
634
    Pipeline * m cancels /mb, yielding total. Rank=rank(total).
    """
    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"]
698
699
700
701
702
703
704
705
706
707
708
        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)
817
818
819
820
821
822
823
824
825
        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"
    )
842
843
844
845
846
847
848
849
850
        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"
    )
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
    FSDP/HSDP all-gather/reduce-scatter happens across d_shard*cp*t ranks.
    This function returns a list of n_level per hierarchy level for that
    shard group, independent of how DP/TP/CP are individually assigned.
    """
    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: Any,
942
943
944
945
946
947
948
949
950

    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
        )
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
        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)
973
974
975
976
977
978
979
980
981
            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: