_utils.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401
  1. # mypy: allow-untyped-defs
  2. import functools
  3. import operator
  4. import re
  5. from collections import deque
  6. from dataclasses import dataclass
  7. from typing import TYPE_CHECKING
  8. from torch.autograd.profiler import profile
  9. from torch.profiler import DeviceType
  10. if TYPE_CHECKING:
  11. from torch.autograd import _KinetoEvent
  12. def _traverse(tree, next_fn, children_fn=lambda x: x.children, reverse: bool = False):
  13. order = reversed if reverse else lambda x: x
  14. remaining = deque(order(tree))
  15. while remaining:
  16. curr_event = next_fn(remaining)
  17. yield curr_event
  18. for child_event in order(children_fn(curr_event)):
  19. remaining.append(child_event)
  20. traverse_dfs = functools.partial(_traverse, next_fn=lambda x: x.pop(), reverse=True)
  21. traverse_bfs = functools.partial(
  22. _traverse, next_fn=lambda x: x.popleft(), reverse=False
  23. )
  24. @dataclass
  25. class EventMetrics:
  26. duration_time_ns: int = 0
  27. self_time_ns: int = 0
  28. idle_time_ns: int = 0
  29. queue_depth: int = 0
  30. @property
  31. def fraction_idle_time(self):
  32. if self.duration_time_ns == 0:
  33. return 0.0
  34. return self.idle_time_ns / self.duration_time_ns
  35. @dataclass
  36. class Interval:
  37. start: int
  38. end: int
  39. queue_depth: int = 0
  40. class EventKey:
  41. def __init__(self, event) -> None:
  42. self.event = event
  43. def __hash__(self):
  44. return hash(self.event.id)
  45. def __eq__(self, other):
  46. return self.event.id == other.event.id
  47. def __repr__(self) -> str:
  48. return f"{self.event.name}"
  49. def intervals_overlap(self, intervals: list[Interval]):
  50. overlap_time = 0
  51. intervals = sorted(intervals, key=lambda x: x.start)
  52. if intervals:
  53. overlap_start = max(self.event.start_time_ns, intervals[0].start)
  54. overlap_end = min(self.event.end_time_ns, intervals[0].end)
  55. if overlap_start < overlap_end:
  56. overlap_time += overlap_end - overlap_start
  57. i, j = 0, 1
  58. while j < len(intervals):
  59. prev_interval = intervals[i]
  60. curr_interval = intervals[j]
  61. j += 1
  62. if prev_interval.end > curr_interval.start:
  63. # Completely subsumed by previous interval
  64. if prev_interval.end > curr_interval.end:
  65. j += 1
  66. continue
  67. else:
  68. curr_interval.start = prev_interval.end
  69. i = j
  70. overlap_start = max(self.event.start_time_ns, curr_interval.start)
  71. overlap_end = min(self.event.end_time_ns, curr_interval.end)
  72. if overlap_start < overlap_end:
  73. overlap_time += overlap_end - overlap_start
  74. return overlap_time
  75. class BasicEvaluation:
  76. def __init__(self, prof: profile) -> None:
  77. self.profile = prof
  78. self.metrics: dict[EventKey, EventMetrics] = {}
  79. self.compute_self_time()
  80. self.event_keys = sorted(
  81. (e for e in self.metrics.keys()), key=lambda x: x.event.start_time_ns
  82. )
  83. self.events = [e.event for e in self.event_keys]
  84. self.cuda_events: list[_KinetoEvent] = []
  85. self.queue_depth_list = self.compute_queue_depth()
  86. self.compute_idle_time()
  87. def compute_self_time(self) -> None:
  88. """
  89. Computes event's self time(total time - time in child ops).
  90. """
  91. assert self.profile.kineto_results is not None
  92. stack = deque(self.profile.kineto_results.experimental_event_tree())
  93. # standard iterating dfs
  94. while stack:
  95. curr_event = stack.pop()
  96. self_time = curr_event.duration_time_ns
  97. for child_event in curr_event.children:
  98. self_time -= child_event.duration_time_ns
  99. stack.append(child_event)
  100. assert EventKey(curr_event) not in self.metrics, (
  101. f"Duplicate id: {curr_event.id}, {curr_event.name}"
  102. )
  103. self.metrics[EventKey(curr_event)] = EventMetrics(self_time_ns=self_time)
  104. self.metrics[
  105. EventKey(curr_event)
  106. ].duration_time_ns = curr_event.duration_time_ns
  107. def compute_queue_depth(self):
  108. """
  109. Computes queue_depth at each event. This will calculate the queue depth data for
  110. All the events in the tree.
  111. This will return a list of Interval of queue depth data of cuda launch and kernels.
  112. """
  113. assert self.profile.kineto_results is not None
  114. cuda_event_list = self.profile.kineto_results.events()
  115. def is_cuda_launch_kernel(e):
  116. """Check if the event is a CUDA launch kernel."""
  117. launch_patterns = {
  118. "cudaLaunchKernel", # Standard CUDA
  119. "cudaLaunchKernelExC", # Extended C
  120. "__cudaLaunchKernel", # Internal
  121. "cudaLaunchCooperativeKernel", # Collaborative (single-device)
  122. "cudaLaunchCooperativeKernelMultiDevice", # Collaborative (multi-devices)
  123. }
  124. name = str(getattr(e, "name", e))
  125. return any(name.startswith(pattern) for pattern in launch_patterns)
  126. def is_cuda_kernel(e):
  127. """Check if the event is a CUDA runtime kernel."""
  128. # Check if the kernel is CUDA
  129. if e.device_type() != DeviceType.CUDA:
  130. return False
  131. name = str(getattr(e, "name", e)).lower()
  132. # Exclude memory operations
  133. exclude_patterns = {"mem", "cpy", "alloc", "free"}
  134. return not any(pattern in name for pattern in exclude_patterns)
  135. cuda_launch_events = sorted(
  136. (e for e in cuda_event_list if is_cuda_launch_kernel(e)),
  137. key=lambda x: x.start_ns(),
  138. )
  139. cuda_kernel_events = sorted(
  140. (e for e in cuda_event_list if is_cuda_kernel(e)),
  141. key=lambda x: x.start_ns(),
  142. )
  143. self.cuda_events = sorted(
  144. cuda_launch_events + cuda_kernel_events, key=lambda x: x.start_ns()
  145. )
  146. kernel_mapping: dict[_KinetoEvent, int] = {}
  147. last_mapped_kernel = 0
  148. for cuda_launch_event in cuda_launch_events:
  149. index = index_of_first_match(
  150. cuda_kernel_events,
  151. lambda x: x.linked_correlation_id()
  152. == cuda_launch_event.linked_correlation_id(),
  153. start=last_mapped_kernel,
  154. )
  155. kernel_mapping[cuda_launch_event] = index
  156. last_mapped_kernel = index if index is not None else last_mapped_kernel
  157. current_kernel_index = 0
  158. spawned_kernel_index = -1
  159. all_events = cuda_launch_events + cuda_kernel_events + self.events
  160. def new_old_event_comparator(event):
  161. if hasattr(event, "start_us"):
  162. return event.start_us() * 1000
  163. if hasattr(event, "start_ns"):
  164. return event.start_ns()
  165. if hasattr(event, "start_time_ns"):
  166. return event.start_time_ns
  167. raise Exception("Unknown Event Type") # noqa: TRY002
  168. queue_depth_list: list[Interval] = []
  169. all_events.sort(key=new_old_event_comparator)
  170. for event in all_events:
  171. # Find latest cuda kernel event
  172. if hasattr(event, "start_us"):
  173. start_time = event.start_us() * 1000
  174. end_time = (event.start_us() + event.duration_us()) * 1000
  175. # Find current spawned cuda kernel event
  176. if event in kernel_mapping and kernel_mapping[event] is not None:
  177. spawned_kernel_index = kernel_mapping[event]
  178. if hasattr(event, "start_ns"):
  179. start_time = event.start_ns()
  180. end_time = event.start_ns() + event.duration_ns()
  181. # Find current spawned cuda kernel event
  182. if event in kernel_mapping and kernel_mapping[event] is not None:
  183. spawned_kernel_index = kernel_mapping[event]
  184. elif hasattr(event, "start_time_ns"):
  185. start_time = event.start_time_ns # type: ignore[attr-defined]
  186. end_time = event.end_time_ns # type: ignore[attr-defined]
  187. while (
  188. current_kernel_index < len(cuda_kernel_events)
  189. and (cuda_kernel_events[current_kernel_index].start_ns()) <= start_time # type: ignore[possibly-undefined]
  190. ):
  191. current_kernel_index += 1
  192. current_queue_depth = spawned_kernel_index - current_kernel_index + 1
  193. current_queue_depth = max(current_queue_depth, 0)
  194. if hasattr(event, "start_us") or hasattr(event, "start_ns"):
  195. queue_depth_list.append(
  196. Interval(start_time, end_time, current_queue_depth) # type: ignore[possibly-undefined]
  197. )
  198. elif hasattr(event, "start_time_ns"):
  199. self.metrics[EventKey(event)].queue_depth = current_queue_depth
  200. return queue_depth_list
  201. def compute_idle_time(self) -> None:
  202. """
  203. Computes idle time of the profile.
  204. """
  205. # Based on queue_depth_list, we can calculate idle time for all the events
  206. idle = False
  207. idle_start = 0
  208. idle_intervals: list[Interval] = []
  209. if self.queue_depth_list and self.events:
  210. idle_intervals += [
  211. Interval(self.events[0].start_time_ns, self.queue_depth_list[0].start),
  212. Interval(self.queue_depth_list[-1].end, self.events[-1].end_time_ns),
  213. ]
  214. for data_point in self.queue_depth_list:
  215. if data_point.queue_depth == 0 and not idle:
  216. idle_start = data_point.end
  217. idle = True
  218. if data_point.queue_depth > 0 and idle:
  219. idle_intervals.append(Interval(idle_start, data_point.start))
  220. idle = False
  221. event_list = [e.event for e in self.metrics.keys()]
  222. for event in event_list:
  223. self.metrics[EventKey(event)].idle_time_ns = EventKey(
  224. event
  225. ).intervals_overlap(idle_intervals)
  226. def rank_events(self, length):
  227. """
  228. Filter and Rank the events based on some heuristics:
  229. 1) Events that are in the falling phase of the queue depth.
  230. 2) Events that have a high idle_time, self_time difference.
  231. Parameters:
  232. length: The number of events to return.
  233. """
  234. # Find the interval when qd is falling to 0
  235. import torch
  236. queue_depth_list = list(reversed(self.queue_depth_list))
  237. qd_values = [e.queue_depth for e in queue_depth_list]
  238. bottom_threashold = 0
  239. top_threashold = 4
  240. decrease_interval = []
  241. i = 0
  242. while i < len(qd_values):
  243. if qd_values[i] > bottom_threashold:
  244. i += 1
  245. continue
  246. for j in range(i + 1, len(qd_values)):
  247. # Find next zero and if the max value between them exceeds
  248. # the threshold, then we have a falling interval
  249. next_minimum_idx = index_of_first_match(
  250. qd_values, lambda x: x <= bottom_threashold, start=j
  251. )
  252. peak_idx = argmax(qd_values, start=j, end=next_minimum_idx)
  253. # if is a valid peak, we add to list and continue
  254. if peak_idx is not None and qd_values[peak_idx] >= top_threashold:
  255. decrease_interval.append(
  256. Interval(
  257. queue_depth_list[peak_idx].start, queue_depth_list[i].start
  258. )
  259. )
  260. i = next_minimum_idx if next_minimum_idx is not None else i
  261. break
  262. i += 1
  263. # Filter out events that are not in the decrease interval
  264. event_list = [
  265. event
  266. for event in self.metrics.keys()
  267. if event.intervals_overlap(decrease_interval)
  268. ]
  269. if event_list:
  270. self_time = torch.tensor(
  271. [self.metrics[event].self_time_ns for event in event_list],
  272. dtype=torch.float32,
  273. )
  274. idle_time = torch.tensor(
  275. [self.metrics[event].fraction_idle_time for event in event_list],
  276. dtype=torch.float32,
  277. )
  278. normalized_gain = (idle_time - torch.mean(idle_time)) / torch.std(idle_time)
  279. normalized_self = (self_time - torch.mean(self_time)) / torch.std(self_time)
  280. heuristic_score_list = normalized_gain + 0.6 * normalized_self
  281. # Sort events by heuristic
  282. event_list = [
  283. event
  284. for _, event in sorted(
  285. zip(heuristic_score_list, event_list),
  286. key=operator.itemgetter(0),
  287. reverse=True,
  288. )
  289. ]
  290. event_list = event_list[:length]
  291. return event_list
  292. def get_optimizable_events(self, length: int = 1, print_enable: bool = True):
  293. event_list = self.rank_events(length)
  294. if not print_enable:
  295. return event_list
  296. output = "Optimizable events:\n" if event_list else "No events to optimize\n"
  297. output += "\n".join(
  298. [
  299. f"""{"-" * 80}
  300. Event: {event}
  301. Source code location: {source_code_location(event.event)}
  302. Percentage idle time: {self.metrics[event].fraction_idle_time * 100:.2f}%
  303. {"-" * 80}"""
  304. for event in event_list
  305. ]
  306. )
  307. if print_enable:
  308. print(output)
  309. return event_list
  310. def index_of_first_match(seq, predicate, start=0, end=None):
  311. if end is None or end >= len(seq):
  312. end = len(seq)
  313. for i in range(start, end):
  314. if predicate(seq[i]):
  315. return i
  316. return None
  317. def argmax(seq, key=lambda x: x, start=0, end=None):
  318. seq = seq[start:end]
  319. if len(seq) == 0:
  320. return None
  321. return seq.index(max(seq, key=key)) + start
  322. def source_code_location(event):
  323. while event is not None:
  324. match = re.search(r"\.py\(.*\)", event.name)
  325. if match is None:
  326. event = event.parent
  327. continue
  328. return event.name
  329. return "No source code location found"
  330. # Provide an OSS workaround for cudagraphs + CUPTI issue
  331. # https://github.com/pytorch/pytorch/issues/75504
  332. # TODO(dberard) - deprecate / remove workaround for CUDA >= 12, when
  333. # we stop supporting older CUDA versions.
  334. def _init_for_cuda_graphs() -> None:
  335. from torch.autograd.profiler import profile
  336. with profile():
  337. pass