Coverage for simulator/utils.py: 75%
161 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2Utilities for the simulator.
3"""
5from __future__ import annotations
7from copy import deepcopy
9import pandas as pd
10import numpy as np
12from scipy.interpolate import interp1d
14from sim_types import ProvisioningResult
15from sim_types import GPUType
16from sim_types import Model
17from sim_types import ModelAllocation
19from typing import Optional
22def to_models_df(
23 models: dict[GPUType, dict[Model, list[ModelAllocation]]]
24) -> pd.DataFrame:
25 """
26 Convert the models dictionary to a pandas DataFrame for easier analysis and visualization.
27 """
28 records = []
29 for gpu_type, model_allocations in models.items():
30 for model, allocations in model_allocations.items():
31 for allocation in allocations:
32 if allocation is None or allocation.get_num_gpus() == 0:
33 continue # Ignoring empty allocations
34 record = {
35 "GPU": gpu_type.value,
36 "Model": model.value,
37 "Devices": allocation.devices,
38 "Replicas": allocation.replicas,
39 "Work": allocation.work,
40 "#GPUs": allocation.get_num_gpus(),
41 "Time (s)": allocation.time,
42 "TTFF (s)": allocation.time_first,
43 "Energy (kWh)": allocation.energy / (60 * 60) / 1000.0, # Convert to kWh
44 "Cost ($)": allocation.cost,
45 }
46 records.append(record)
47 df = pd.DataFrame(records)
48 df = df.set_index(["GPU", "Model"])
49 df = df.round(2)
51 total = df.sum(numeric_only=True)
52 total["Time (s)"] = df["Time (s)"].groupby(level="Model").max().sum()
53 total["TTFF (s)"] = df["TTFF (s)"].groupby(level="Model").min().sum()
54 total.name = ("TOTAL", "")
55 df = pd.concat([df, total.to_frame().T])
57 df[["Devices", "Replicas", "#GPUs", "Work"]] = df[["Devices", "Replicas", "#GPUs", "Work"]].astype(int)
59 return df
62def coalesce_models(
63 models: dict[GPUType, dict[Model, list[ModelAllocation]]]
64) -> dict[GPUType, dict[Model, list[ModelAllocation]]]:
65 """The models with the same parallelism and same work, should be accounted as replicas."""
66 merged: dict[GPUType, dict[Model, list[ModelAllocation]]] = {}
67 for gpu_type, model_dict in models.items():
68 merged[gpu_type] = {}
69 for model_name, allocations in model_dict.items():
70 merged_allocations: list[ModelAllocation] = []
71 for alloc in allocations:
72 # Check if there's an existing allocation with the same devices and work
73 match = next((
74 model_alloc
75 for model_alloc in merged_allocations
76 if model_alloc.devices == alloc.devices and model_alloc.work == alloc.work
77 ), None)
78 if match:
79 # If found, increment replicas and aggregate energy/cost
80 match.replicas += 1
81 match.energy += alloc.energy
82 match.cost += alloc.cost
83 else:
84 # Otherwise, add as new allocation
85 merged_allocations.append(deepcopy(alloc))
86 merged[gpu_type][model_name] = merged_allocations
87 return merged
90def simplify_model_allocations(
91 models: dict[GPUType, dict[Model, list[ModelAllocation]]],
92) -> dict[GPUType, dict[Model, list[ModelAllocation]]]:
93 """
94 Simplify model allocations by merging replicas with the same number of devices.
95 This is to reduce the search space for the optimization loop.
96 """
97 new_models = deepcopy(models)
98 for gpu_type in new_models.keys():
99 for model in new_models[gpu_type].keys():
100 model_instances = new_models[gpu_type][model]
101 alloc_map: dict[int, ModelAllocation] = {}
102 for model_instance in model_instances:
103 if model_instance.get_num_gpus() == 0:
104 continue
105 if model_instance.devices not in alloc_map:
106 alloc_map[model_instance.devices] = deepcopy(model_instance)
107 else:
108 alloc_map[model_instance.devices].replicas += model_instance.replicas
109 new_models[gpu_type][model] = list(alloc_map.values())
110 return new_models
113def find_fastest_provisioning(
114 provisioning: ProvisioningResult,
115) -> int:
116 """Find the fastest provisioning option."""
117 min_latency = min(provisioning.latencies)
118 min_latency_index = provisioning.latencies.index(min_latency)
119 return min_latency_index
122def find_fastest_ttff_provisioning(
123 provisioning: ProvisioningResult,
124) -> int:
125 """Find the fastest provisioning option."""
126 min_ttff = min(provisioning.ttffs)
127 min_ttff_index = provisioning.ttffs.index(min_ttff)
128 return min_ttff_index
131def find_cheapest_provisioning(
132 provisioning: ProvisioningResult,
133) -> int:
134 """Find the cheapest provisioning option."""
135 min_cost = min(provisioning.costs)
136 min_cost_index = provisioning.costs.index(min_cost)
137 return min_cost_index
140def find_most_cost_effective_provisioning(
141 provisioning: ProvisioningResult,
142) -> int:
143 """Find the most cost-effective provisioning option."""
144 min_cost = min(provisioning.costs)
145 min_latency = min(provisioning.latencies)
146 min_cost_index = provisioning.costs.index(min_cost)
147 min_latency_index = provisioning.latencies.index(min_latency)
148 if min_cost_index == min_latency_index:
149 return min_cost_index
151 # if the indices are different, return the provisioning option with the minimum cost*latency
152 cost_latency_list = [
153 cost * latency
154 for cost, latency in zip(provisioning.costs, provisioning.latencies)
155 ]
156 min_cost_latency = min(cost_latency_list)
157 min_cost_latency_index = cost_latency_list.index(min_cost_latency)
158 return min_cost_latency_index
161def find_most_energy_efficient_provisioning(
162 provisioning: ProvisioningResult,
163) -> int:
164 """Find the most energy-efficient provisioning option."""
165 min_energy = min(provisioning.energies)
166 min_latency = min(provisioning.latencies)
167 min_energy_index = provisioning.energies.index(min_energy)
168 min_latency_index = provisioning.latencies.index(min_latency)
169 if min_energy_index == min_latency_index:
170 return min_energy_index
172 # if the indices are different, return the provisioning option with the minimum energy*latency
173 energy_latency_list = [
174 energy * latency
175 for energy, latency in zip(provisioning.energies, provisioning.latencies)
176 ]
177 min_energy_latency = min(energy_latency_list)
178 min_energy_latency_index = energy_latency_list.index(min_energy_latency)
179 return min_energy_latency_index
182def find_pareto_frontier(
183 latency_list: list[float],
184 energy_list: list[float],
185 provision: list[float]
186) -> tuple[list[float], list[float], list[float]]:
187 pareto_provision = []
188 pareto_latency = []
189 pareto_energy = []
190 for i in range(len(latency_list)):
191 dominated = False
192 for j in range(len(latency_list)):
193 if i != j:
194 if latency_list[j] <= latency_list[i] and energy_list[j] <= energy_list[i]:
195 if latency_list[j] < latency_list[i] or energy_list[j] < energy_list[i]:
196 dominated = True
197 break
198 if not dominated:
199 pareto_provision.append(provision[i])
200 pareto_latency.append(latency_list[i])
201 pareto_energy.append(energy_list[i])
202 return pareto_provision, pareto_latency, pareto_energy
205def get_pareto_frontier_paper(
206 points: np.ndarray,
207 max_y: Optional[float] = None,
208 max_x: Optional[float] = None,
209) -> np.ndarray:
210 """
211 Calculate the Pareto frontier from a set of data points
212 """
213 if points.size == 0:
214 return points.copy()
216 # points = points[np.argsort(points[:, 0])]
217 points = points[np.lexsort((points[:, 1], points[:, 0]))]
219 pareto_front = [points[0]]
220 for point in points[1:]:
221 if point[1] < pareto_front[-1][1]:
222 pareto_front.append(point)
224 # Add extreme points to the Pareto frontier
225 extreme_point_0 = [pareto_front[0][0], max(points[:, 1])]
226 extreme_point_1 = [max(points[:, 0]), pareto_front[-1][1]]
227 pareto_front.append(extreme_point_0)
228 pareto_front.append(extreme_point_1)
230 if max_x is not None:
231 candidate = np.array([max_x, min(points[:, 1])])
232 if candidate[0] > pareto_front[-1][0] and candidate[1] <= pareto_front[-1][1]:
233 pareto_front.append(candidate)
234 if max_y is not None:
235 candidate = np.array([min(points[:, 0]), max_y])
236 if candidate[1] > pareto_front[0][1] and candidate[0] <= pareto_front[0][0]:
237 pareto_front.append(candidate)
239 pareto_front_np = np.array(pareto_front)
240 pareto_front_np = pareto_front_np[np.lexsort((
241 -pareto_front_np[:, 1],
242 pareto_front_np[:, 0]))]
244 # Avoid repeated points
245 _, idx = np.unique(pareto_front_np, axis=0, return_index=True)
246 pareto_front_np = pareto_front_np[np.sort(idx)]
248 return pareto_front_np
251def get_pareto_frontier(
252 ttff_list: list[float],
253 costs: list[float],
254 max_y: Optional[float] = None,
255 max_x: Optional[float] = None,
256) -> np.ndarray:
257 points = np.array(list(zip(ttff_list, costs)))
258 return get_pareto_frontier_paper(
259 points,
260 max_x,
261 max_y,
262 )
265def clean_frontier(
266 frontier: np.ndarray
267) -> np.ndarray:
268 F = frontier[np.argsort(frontier[:, 0])]
269 xs = []
270 ys = []
271 i = 0
272 while i < len(F):
273 x = F[i, 0]
274 same_x = F[F[:, 0] == x]
275 xs.append(x)
276 ys.append(same_x[:, 1].min())
277 i += len(same_x)
278 return np.column_stack([xs, ys])
281def area_between_frontiers(
282 A: np.ndarray,
283 B: np.ndarray,
284 n: int = 5000
285) -> np.ndarray:
286 A = clean_frontier(A)
287 B = clean_frontier(B)
288 xmin = max(A[:, 0].min(), B[:, 0].min())
289 xmax = min(A[:, 0].max(), B[:, 0].max())
290 xs = np.linspace(xmin, xmax, n)
291 fA = interp1d(A[:, 0], A[:, 1], kind="linear")
292 fB = interp1d(B[:, 0], B[:, 1], kind="linear")
293 yA = fA(xs)
294 yB = fB(xs)
295 # return np.trapezoid(yB - yA, xs)
296 delta = yB - yA
297 return 100.0 * delta / yB