__init__.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425
  1. # mypy: allow-untyped-defs
  2. r"""
  3. This package enables an interface for accessing MTIA backend in python
  4. """
  5. import threading
  6. import warnings
  7. from typing import Any, Callable, Optional, Union
  8. import torch
  9. from torch import device as _device, Tensor
  10. from torch._utils import _dummy_type, _LazySeedTracker, classproperty
  11. from torch.types import Device
  12. from ._utils import _get_device_index
  13. _device_t = Union[_device, str, int]
  14. # torch.mtia.Event/Stream is alias of torch.Event/Stream
  15. Event = torch.Event
  16. Stream = torch.Stream
  17. _initialized = False
  18. _queued_calls: list[
  19. tuple[Callable[[], None], list[str]]
  20. ] = [] # don't invoke these until initialization occurs
  21. _tls = threading.local()
  22. _initialization_lock = threading.Lock()
  23. _lazy_seed_tracker = _LazySeedTracker()
  24. if hasattr(torch._C, "_mtia_exchangeDevice"):
  25. _exchange_device = torch._C._mtia_exchangeDevice
  26. else:
  27. def _exchange_device(device: int) -> int:
  28. if device < 0:
  29. return -1
  30. raise RuntimeError("PyTorch was compiled without MTIA support")
  31. if hasattr(torch._C, "_mtia_maybeExchangeDevice"):
  32. _maybe_exchange_device = torch._C._mtia_maybeExchangeDevice
  33. else:
  34. def _maybe_exchange_device(device: int) -> int:
  35. if device < 0:
  36. return -1
  37. raise RuntimeError("PyTorch was compiled without MTIA support")
  38. def init():
  39. _lazy_init()
  40. def is_initialized():
  41. r"""Return whether PyTorch's MTIA state has been initialized."""
  42. return _initialized and not _is_in_bad_fork()
  43. def _is_in_bad_fork() -> bool:
  44. return torch._C._mtia_isInBadFork()
  45. def _lazy_init() -> None:
  46. global _initialized, _queued_calls
  47. if is_initialized() or hasattr(_tls, "is_initializing"):
  48. return
  49. with _initialization_lock:
  50. # We be double-checking locking, boys! This is OK because
  51. # the above test was GIL protected anyway. The inner test
  52. # is for when a thread blocked on some other thread which was
  53. # doing the initialization; when they get the lock, they will
  54. # find there is nothing left to do.
  55. if is_initialized():
  56. return
  57. # It is important to prevent other threads from entering _lazy_init
  58. # immediately, while we are still guaranteed to have the GIL, because some
  59. # of the C calls we make below will release the GIL
  60. if _is_in_bad_fork():
  61. raise RuntimeError(
  62. "Cannot re-initialize MTIA in forked subprocess. To use MTIA with "
  63. "multiprocessing, you must use the 'spawn' start method"
  64. )
  65. if not _is_compiled():
  66. raise AssertionError(
  67. "Torch not compiled with MTIA enabled. "
  68. "Ensure you have `import mtia.host_runtime.torch_mtia.dynamic_library` in your python "
  69. "src file and include `//mtia/host_runtime/torch_mtia:torch_mtia` as "
  70. "your target dependency!"
  71. )
  72. torch._C._mtia_init()
  73. # Some of the queued calls may reentrantly call _lazy_init();
  74. # we need to just return without initializing in that case.
  75. # However, we must not let any *other* threads in!
  76. _tls.is_initializing = True
  77. _queued_calls.extend(calls for calls in _lazy_seed_tracker.get_calls() if calls)
  78. try:
  79. for queued_call, orig_traceback in _queued_calls:
  80. try:
  81. queued_call()
  82. except Exception as e:
  83. msg = (
  84. f"MTIA call failed lazily at initialization with error: {str(e)}\n\n"
  85. f"MTIA call was originally invoked at:\n\n{''.join(orig_traceback)}"
  86. )
  87. raise DeferredMtiaCallError(msg) from e
  88. finally:
  89. delattr(_tls, "is_initializing")
  90. _initialized = True
  91. class DeferredMtiaCallError(Exception):
  92. pass
  93. def _is_compiled() -> bool:
  94. r"""Return true if compiled with MTIA support."""
  95. return torch._C._mtia_isBuilt()
  96. def is_available() -> bool:
  97. r"""Return true if MTIA device is available"""
  98. if not _is_compiled():
  99. return False
  100. # MTIA has to init devices first to know if there is any devices available.
  101. return device_count() > 0
  102. def synchronize(device: Optional[_device_t] = None) -> None:
  103. r"""Waits for all jobs in all streams on a MTIA device to complete."""
  104. with torch.mtia.device(device):
  105. return torch._C._mtia_deviceSynchronize()
  106. def device_count() -> int:
  107. r"""Return the number of MTIA devices available."""
  108. # TODO: Update _accelerator_hooks_device_count to abstract a MTIA device count API
  109. return torch._C._mtia_getDeviceCount()
  110. def current_device() -> int:
  111. r"""Return the index of a currently selected device."""
  112. return torch._C._accelerator_hooks_get_current_device()
  113. def current_stream(device: Optional[_device_t] = None) -> Stream:
  114. r"""Return the currently selected :class:`Stream` for a given device.
  115. Args:
  116. device (torch.device or int, optional): selected device. Returns
  117. the currently selected :class:`Stream` for the current device, given
  118. by :func:`~torch.mtia.current_device`, if :attr:`device` is ``None``
  119. (default).
  120. """
  121. return torch._C._mtia_getCurrentStream(_get_device_index(device, optional=True))
  122. def default_stream(device: Optional[_device_t] = None) -> Stream:
  123. r"""Return the default :class:`Stream` for a given device.
  124. Args:
  125. device (torch.device or int, optional): selected device. Returns
  126. the default :class:`Stream` for the current device, given by
  127. :func:`~torch.mtia.current_device`, if :attr:`device` is ``None``
  128. (default).
  129. """
  130. return torch._C._mtia_getDefaultStream(_get_device_index(device, optional=True))
  131. def record_memory_history(
  132. enabled: Optional[str] = "all", stacks: str = "python", max_entries: int = 0
  133. ) -> None:
  134. r"""Enable/Disable the memory profiler on MTIA allocator
  135. Args:
  136. enabled (all or state, optional) selected device. Returns
  137. statistics for the current device, given by current_device(),
  138. if device is None (default).
  139. stacks ("python" or "cpp", optional). Select the stack trace to record.
  140. max_entries (int, optional). Maximum number of entries to record.
  141. """
  142. if not is_initialized():
  143. return
  144. torch._C._mtia_recordMemoryHistory(enabled, stacks, max_entries)
  145. def snapshot() -> dict[str, Any]:
  146. r"""Return a dictionary of MTIA memory allocator history"""
  147. return torch._C._mtia_memorySnapshot()
  148. def attach_out_of_memory_observer(
  149. observer: Callable[[int, int, int, int], None],
  150. ) -> None:
  151. r"""Attach an out-of-memory observer to MTIA memory allocator"""
  152. torch._C._mtia_attachOutOfMemoryObserver(observer)
  153. def is_bf16_supported(including_emulation: bool = True):
  154. return True
  155. def get_device_capability(device: Optional[_device_t] = None) -> tuple[int, int]:
  156. r"""Return capability of a given device as a tuple of (major version, minor version).
  157. Args:
  158. device (torch.device or int, optional) selected device. Returns
  159. statistics for the current device, given by current_device(),
  160. if device is None (default).
  161. """
  162. return torch._C._mtia_getDeviceCapability(_get_device_index(device, optional=True))
  163. def empty_cache() -> None:
  164. r"""Empty the MTIA device cache."""
  165. return torch._C._mtia_emptyCache()
  166. def set_stream(stream: Stream):
  167. r"""Set the current stream.This is a wrapper API to set the stream.
  168. Usage of this function is discouraged in favor of the ``stream``
  169. context manager.
  170. Args:
  171. stream (Stream): selected stream. This function is a no-op
  172. if this argument is ``None``.
  173. """
  174. if stream is None:
  175. return
  176. torch._C._mtia_setCurrentStream(stream)
  177. def set_device(device: _device_t) -> None:
  178. r"""Set the current device.
  179. Args:
  180. device (torch.device or int): selected device. This function is a no-op
  181. if this argument is negative.
  182. """
  183. device = _get_device_index(device)
  184. if device >= 0:
  185. torch._C._accelerator_hooks_set_current_device(device)
  186. def get_device_properties(device: Optional[_device_t] = None) -> dict[str, Any]:
  187. r"""Return a dictionary of MTIA device properties
  188. Args:
  189. device (torch.device or int, optional) selected device. Returns
  190. statistics for the current device, given by current_device(),
  191. if device is None (default).
  192. """
  193. return torch._C._mtia_getDeviceProperties(_get_device_index(device, optional=True))
  194. class device:
  195. r"""Context-manager that changes the selected device.
  196. Args:
  197. device (torch.device or int): device index to select. It's a no-op if
  198. this argument is a negative integer or ``None``.
  199. """
  200. def __init__(self, device: Any):
  201. self.idx = _get_device_index(device, optional=True)
  202. self.prev_idx = -1
  203. def __enter__(self):
  204. self.prev_idx = torch._C._accelerator_hooks_maybe_exchange_device(self.idx)
  205. def __exit__(self, type: Any, value: Any, traceback: Any):
  206. self.idx = torch._C._accelerator_hooks_maybe_exchange_device(self.prev_idx)
  207. return False
  208. class StreamContext:
  209. r"""Context-manager that selects a given stream.
  210. All MTIA kernels queued within its context will be enqueued on a selected
  211. stream.
  212. Args:
  213. Stream (Stream): selected stream. This manager is a no-op if it's
  214. ``None``.
  215. .. note:: Streams are per-device.
  216. """
  217. cur_stream: Optional["torch.mtia.Stream"]
  218. def __init__(self, stream: Optional["torch.mtia.Stream"]):
  219. self.cur_stream = None
  220. self.stream = stream
  221. self.idx = _get_device_index(None, True)
  222. if not torch.jit.is_scripting():
  223. if self.idx is None:
  224. self.idx = -1
  225. self.src_prev_stream = (
  226. None if not torch.jit.is_scripting() else torch.mtia.default_stream(None)
  227. )
  228. self.dst_prev_stream = (
  229. None if not torch.jit.is_scripting() else torch.mtia.default_stream(None)
  230. )
  231. def __enter__(self):
  232. # Local cur_stream variable for type refinement
  233. cur_stream = self.stream
  234. # Return if stream is None or MTIA device not available
  235. if cur_stream is None or self.idx == -1:
  236. return
  237. self.src_prev_stream = torch.mtia.current_stream(None)
  238. # If the stream is not on the current device, then
  239. # set the current stream on the device
  240. if self.src_prev_stream.device != cur_stream.device:
  241. with device(cur_stream.device):
  242. self.dst_prev_stream = torch.mtia.current_stream(cur_stream.device)
  243. torch.mtia.set_stream(cur_stream)
  244. def __exit__(self, type: Any, value: Any, traceback: Any):
  245. # Local cur_stream variable for type refinement
  246. cur_stream = self.stream
  247. # If stream is None or no MTIA device available, return
  248. if cur_stream is None or self.idx == -1:
  249. return
  250. # Reset the stream on the original device
  251. # and destination device
  252. if self.src_prev_stream.device != cur_stream.device: # type: ignore[union-attr]
  253. torch.mtia.set_stream(self.dst_prev_stream) # type: ignore[arg-type]
  254. torch.mtia.set_stream(self.src_prev_stream) # type: ignore[arg-type]
  255. def _set_stream_by_id(stream_id, device_index, device_type):
  256. r"""set stream specified by the stream id, device index and
  257. device type
  258. Args: stream_id (int): stream id in stream pool
  259. device_index (int): device index in topo
  260. device_type (int): enum device type
  261. """
  262. torch._C._mtia_setStream(stream_id, device_index, device_type)
  263. def stream(stream: Optional["torch.mtia.Stream"]) -> StreamContext:
  264. r"""Wrap around the Context-manager StreamContext that selects a given stream.
  265. Arguments:
  266. stream (Stream): selected stream. This manager is a no-op if it's
  267. ``None``.
  268. .. note:: In eager mode stream is of type Stream class while in JIT it doesn't support torch.mtia.stream
  269. """
  270. return StreamContext(stream)
  271. def get_rng_state(device: Union[int, str, torch.device] = "mtia") -> Tensor:
  272. r"""Returns the random number generator state as a ByteTensor.
  273. Args:
  274. device (torch.device or int, optional): The device to return the RNG state of.
  275. Default: ``'mtia'`` (i.e., ``torch.device('mtia')``, the current mtia device).
  276. """
  277. warnings.warn(
  278. "get_rng_state is not implemented in torch.mtia",
  279. UserWarning,
  280. stacklevel=2,
  281. )
  282. return torch.zeros([1], dtype=torch.uint8, device=device)
  283. def set_rng_state(
  284. new_state: Tensor, device: Union[int, str, torch.device] = "mtia"
  285. ) -> None:
  286. r"""Sets the random number generator state.
  287. Args:
  288. new_state (torch.ByteTensor): The desired state
  289. device (torch.device or int, optional): The device to set the RNG state.
  290. Default: ``'mtia'`` (i.e., ``torch.device('mtia')``, the current mtia device).
  291. """
  292. warnings.warn(
  293. "set_rng_state is not implemented in torch.mtia",
  294. UserWarning,
  295. stacklevel=2,
  296. )
  297. from .memory import * # noqa: F403
  298. __all__ = [
  299. "init",
  300. "is_available",
  301. "is_initialized",
  302. "synchronize",
  303. "device_count",
  304. "current_device",
  305. "current_stream",
  306. "default_stream",
  307. "memory_stats",
  308. "max_memory_allocated",
  309. "memory_allocated",
  310. "reset_peak_memory_stats",
  311. "get_device_capability",
  312. "get_device_properties",
  313. "record_memory_history",
  314. "snapshot",
  315. "attach_out_of_memory_observer",
  316. "empty_cache",
  317. "set_device",
  318. "set_stream",
  319. "stream",
  320. "device",
  321. "set_rng_state",
  322. "get_rng_state",
  323. "is_bf16_supported",
  324. ]