Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / sapp_nd / nd / ratios.py: 98%
82 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Ratios that correct ND's parts from a measured round, in the file ``run_nd -c`` reads.
17A round of strategies measured on a cluster and classified into the parts ND
18estimates (``nd.trace_classify``) gives, for each part, the ratio of what the
19cluster measured to what ND estimated, summed over the round. ND reads the
20ratios from a file with ``-c`` (``estimate.apply_regression_coefficients``)
21and re-scores every candidate with them: each part times its ratio, summed,
22in milliseconds.
24``run_nd --real_csv <csv> --write_ratios <json>`` fits them on a comparison,
25writes the file, and prints how well they predict each measured strategy
26when fitted on the others.
27"""
28import json
29from typing import Any, Dict, List, Optional, Sequence, Tuple
31from hyper_parallel.auto_parallel.sapp_nd.nd.debug import PerfParts, TRAINER_STEP
33# Each ratio of the file: the ND parts it scales and the measured columns it
34# is fitted to. The classifier's Ascend path files FSDP's gathers and
35# reduce-scatters under op_wait and the all-reduces under dp_wait, and real
36# data cannot tell the bubble from P2P time, so one ratio covers both.
37FITS: Tuple[Tuple[str, Tuple[PerfParts, ...], Tuple[str, ...]], ...] = (
38 ("COMPUTE", (PerfParts.FW_COMPUTE, PerfParts.BW_COMPUTE, PerfParts.RECOMPUTE), ("comp",)),
39 ("DP_COMM", (PerfParts.DP_COMM,), ("op_wait",)),
40 ("DP_REDUCE", (PerfParts.DP_REDUCE,), ("dp_wait",)),
41 ("MP_COMM", (PerfParts.MP_COMM,), ("mp_wait", "sp_wait")),
42 ("EP_COMM", (PerfParts.EP_COMM,), ("ep_wait",)),
43 ("CP_COMM", (PerfParts.CP_COMM,), ("cp_wait",)),
44 ("PP_COMM", (PerfParts.PP_COMM, PerfParts.BUBBLE), ("pp_wait",)),
45)
47# The parts a score splits into, in the order a comparison lists them.
48SCORE_PARTS = [part for part in PerfParts if part not in {PerfParts.TOTAL, PerfParts.MEMORY}]
50# The share of ND's whole estimate below which a part is not priced at all:
51# a bubble without a pipeline leaves a rounding residue of 1e-16 of it.
52NEGLIGIBLE = 1e-9
55def label(config: Any) -> str:
56 """A strategy's degrees above 1, DP always: ``DP 64 EP 2 OP 32``."""
57 return " ".join(f"{dim} {value}" for dim, value in zip(config.keys(), config.values())
58 if value not in ("1", "False") or str(dim) == "DP")
61def _estimated(entry: tuple) -> Dict[PerfParts, float]:
62 """ND's parts of one comparison entry, by part."""
63 return dict(zip(SCORE_PARTS, entry[4]))
66def _measured(entry: tuple, columns: Sequence[str]) -> float:
67 """The milliseconds one comparison entry measured in *columns*."""
68 return sum(entry[5].get(column) or 0.0 for column in columns)
71def busy(entry: tuple) -> float:
72 """The measured step of a comparison entry less idle: every column a ratio is fitted to."""
73 return sum(_measured(entry, columns) for _, _, columns in FITS)
76def step(entry: tuple) -> float:
77 """The measured step of a comparison entry, the trainer's own where it has one.
79 ``entry[2]`` is the mean of the profiled steps, and the profiler is not free:
80 it costs 0.33 to 0.84 s a step at EP 1 against 0.01 to 0.10 at EP 8, so
81 ranking on it reversed EP 1 against EP 2 on the 2 October round. Where the
82 round harvested the trainer's own step time the comparison ranks on that
83 instead, while every part, and with it ``busy`` above, stays as the profile
84 measured it.
85 """
86 return entry[5].get(TRAINER_STEP) or entry[2]
89def fit_ratios(entries: Sequence[tuple]) -> Dict[str, float]:
90 """Fit one ratio per part, measured over estimated, summed over *entries*.
92 A part ND prices at next to nothing over the round, TP where no strategy
93 splits tensors or the rounding residue of a bubble without a pipeline,
94 takes compute's ratio: it then only turns ND's units into milliseconds, as
95 ND weighs every part in one unit. The bubble takes P2P's.
97 Args:
98 entries: ``(config, peak_mem, real_time, score, parts, real_parts)``, as
99 ``ParallelizeLayer.compare_with_csv`` returns them.
101 Returns:
102 The ratio of each key ``apply_regression_coefficients`` reads.
104 Raises:
105 ValueError: When ND prices no compute over the round, the one part every
106 ratio can fall back on.
107 """
108 fitted: Dict[str, Optional[float]] = {}
109 whole = sum(sum(_estimated(entry).values()) for entry in entries)
110 for key, parts, columns in FITS:
111 estimated = sum(sum(_estimated(entry)[part] for part in parts) for entry in entries)
112 measured = sum(_measured(entry, columns) for entry in entries)
113 fitted[key] = measured / estimated if estimated > NEGLIGIBLE * whole else None
114 unit = fitted["COMPUTE"]
115 if unit is None:
116 raise ValueError("ND prices no compute over the round: no ratio can be fitted")
117 ratios = {key: unit if value is None else value for key, value in fitted.items()}
118 ratios[PerfParts.BUBBLE.name] = ratios[PerfParts.PP_COMM.name]
119 return ratios
122def corrected(parts: Dict[PerfParts, float], ratios: Dict[str, float]) -> float:
123 """ND's estimate of one strategy with *ratios*, in milliseconds, as ``run_nd -c`` scores it."""
124 total = 0.0
125 for key, keyed_parts, _ in FITS:
126 for part in keyed_parts:
127 ratio = ratios[PerfParts.BUBBLE.name] if part == PerfParts.BUBBLE else ratios[key]
128 total += (parts.get(part) or 0.0) * ratio
129 return total
132def leave_one_out(entries: Sequence[tuple]) -> List[Tuple[tuple, Optional[float]]]:
133 """Each entry with ND's estimate of it corrected by ratios fitted on the others, None alone."""
134 entries = list(entries)
135 if len(entries) < 2:
136 return [(entry, None) for entry in entries]
137 return [(entry, corrected(_estimated(entry), fit_ratios(entries[:i] + entries[i + 1:])))
138 for i, entry in enumerate(entries)]
141def write_ratios(path: str, ratios: Dict[str, float], entries: Sequence[tuple], source: str) -> None:
142 """Write *ratios* where ``run_nd -c`` reads them, with the round they were fitted on."""
143 record = dict(ratios)
144 record["_fitted_on"] = {"csv": source, "strategies": [label(entry[0]) for entry in entries]}
145 with open(path, "w", encoding="utf-8") as handle:
146 json.dump(record, handle, indent=2)
149def _ranks(values: Sequence[float]) -> List[float]:
150 """1-based ranks of *values*, a tie sharing the mean of its ranks."""
151 order = sorted(range(len(values)), key=lambda index: values[index])
152 ranks = [0.0] * len(values)
153 start = 0
154 while start < len(order):
155 end = start
156 while end + 1 < len(order) and values[order[end + 1]] == values[order[start]]:
157 end += 1
158 for position in range(start, end + 1):
159 ranks[order[position]] = (start + end) / 2 + 1
160 start = end + 1
161 return ranks
164def spearman(first: Sequence[float], second: Sequence[float]) -> Optional[float]:
165 """Rank correlation of two series, None when it is undefined."""
166 if len(first) < 3:
167 return None
168 ranks_a, ranks_b = _ranks(first), _ranks(second)
169 mean = (len(first) + 1) / 2
170 covariance = sum((a - mean) * (b - mean) for a, b in zip(ranks_a, ranks_b))
171 spread = (sum((a - mean) ** 2 for a in ranks_a) * sum((b - mean) ** 2 for b in ranks_b)) ** 0.5
172 return covariance / spread if spread else None
175def report(entries: Sequence[tuple], ratios: Dict[str, float]) -> List[str]:
176 """The ratios, each measured strategy corrected in and out of sample, and the corrected ND's pick.
178 The corrected estimate leaves out idle, which ND does not price, so it is
179 set beside the busy time; the cost of following the corrected ND is
180 measured on the whole step, the trainer's own where the round harvested it
181 and the profiled one otherwise. The last line says which, because the two
182 disagree by enough to change the answer.
183 """
184 lines = ["Ratios, measured milliseconds per ND unit:"]
185 lines += [f" {key:10s} {value:.4e}" for key, value in ratios.items()]
186 rows = sorted(((corrected(_estimated(entry), ratios), out, entry) for entry, out in leave_one_out(entries)),
187 key=lambda row: row[0])
188 lines.append(f" {'strategy':22s} {'corrected':>9s} {'left out':>9s} {'busy':>8s} {'error':>7s} {'step':>8s}")
189 for estimate, out, entry in rows:
190 held = f"{out:9.1f} {busy(entry):8.1f} {out / busy(entry) - 1:+7.1%}" if out is not None else \
191 f"{'-':>9s} {busy(entry):8.1f} {'-':>7s}"
192 lines.append(f" {label(entry[0]):22s} {estimate:9.1f} {held} {step(entry):8.1f}")
193 pick, fastest = rows[0][2], min(entries, key=step)
194 lines.append(f"Corrected, ND ranks {label(pick[0])} first: it measures {step(pick):.1f} ms; the fastest, "
195 f"{label(fastest[0])}, {step(fastest):.1f}: following the corrected ND costs "
196 f"{step(pick) / step(fastest) - 1:.1%}.")
197 correlation = spearman([estimate for estimate, _, _ in rows], [step(entry) for _, _, entry in rows])
198 if correlation is not None:
199 lines.append(f"Rank correlation of the corrected estimate with the measured step: {correlation:+.2f}.")
200 harvested = sum(1 for entry in entries if entry[5].get(TRAINER_STEP))
201 lines.append(f"The step above is the trainer's own on {harvested} of {len(entries)} strategies and the mean "
202 f"of the profiled steps on the rest; every part is the profile's either way.")
203 return lines