streams.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. import ctypes
  2. import torch
  3. from ._utils import _dummy_type
  4. if not hasattr(torch._C, '_CudaStreamBase'):
  5. # Define dummy base classes
  6. torch._C.__dict__['_CudaStreamBase'] = _dummy_type('_CudaStreamBase')
  7. torch._C.__dict__['_CudaEventBase'] = _dummy_type('_CudaEventBase')
  8. class Stream(torch._C._CudaStreamBase):
  9. r"""Wrapper around a CUDA stream.
  10. A CUDA stream is a linear sequence of execution that belongs to a specific
  11. device, independent from other streams. See :ref:`cuda-semantics` for
  12. details.
  13. Args:
  14. device(torch.device or int, optional): a device on which to allocate
  15. the stream. If :attr:`device` is ``None`` (default) or a negative
  16. integer, this will use the current device.
  17. priority(int, optional): priority of the stream. Can be either
  18. -1 (high priority) or 0 (low priority). By default, streams have
  19. priority 0.
  20. .. note:: Although CUDA versions >= 11 support more than two levels of
  21. priorities, in PyTorch, we only support two levels of priorities.
  22. """
  23. def __new__(cls, device=None, priority=0, **kwargs):
  24. # setting device manager is expensive, so we avoid it unless necessary
  25. if device is None or ("stream_id" in kwargs and "device_index" in kwargs):
  26. return super(Stream, cls).__new__(cls, priority=priority, **kwargs)
  27. else:
  28. with torch.cuda.device(device):
  29. return super(Stream, cls).__new__(cls, priority=priority, **kwargs)
  30. def wait_event(self, event):
  31. r"""Makes all future work submitted to the stream wait for an event.
  32. Args:
  33. event (torch.cuda.Event): an event to wait for.
  34. .. note:: This is a wrapper around ``cudaStreamWaitEvent()``: see
  35. `CUDA Stream documentation`_ for more info.
  36. This function returns without waiting for :attr:`event`: only future
  37. operations are affected.
  38. .. _CUDA Stream documentation:
  39. https://docs.nvidia.com/cuda/cuda-runtime-api/group__CUDART__STREAM.html
  40. """
  41. event.wait(self)
  42. def wait_stream(self, stream):
  43. r"""Synchronizes with another stream.
  44. All future work submitted to this stream will wait until all kernels
  45. submitted to a given stream at the time of call complete.
  46. Args:
  47. stream (Stream): a stream to synchronize.
  48. .. note:: This function returns without waiting for currently enqueued
  49. kernels in :attr:`stream`: only future operations are affected.
  50. """
  51. self.wait_event(stream.record_event())
  52. def record_event(self, event=None):
  53. r"""Records an event.
  54. Args:
  55. event (torch.cuda.Event, optional): event to record. If not given, a new one
  56. will be allocated.
  57. Returns:
  58. Recorded event.
  59. """
  60. if event is None:
  61. event = Event()
  62. event.record(self)
  63. return event
  64. def query(self):
  65. r"""Checks if all the work submitted has been completed.
  66. Returns:
  67. A boolean indicating if all kernels in this stream are completed."""
  68. return super().query()
  69. def synchronize(self):
  70. r"""Wait for all the kernels in this stream to complete.
  71. .. note:: This is a wrapper around ``cudaStreamSynchronize()``: see
  72. `CUDA Stream documentation`_ for more info.
  73. """
  74. super().synchronize()
  75. @property
  76. def _as_parameter_(self):
  77. return ctypes.c_void_p(self.cuda_stream)
  78. def __eq__(self, o):
  79. if isinstance(o, Stream):
  80. return super().__eq__(o)
  81. return False
  82. def __hash__(self):
  83. return hash((self.cuda_stream, self.device))
  84. def __repr__(self):
  85. return ('<torch.cuda.Stream device={0} cuda_stream={1:#x}>'
  86. .format(self.device, self.cuda_stream))
  87. class ExternalStream(Stream):
  88. r"""Wrapper around an externally allocated CUDA stream.
  89. This class is used to wrap streams allocated in other libraries in order
  90. to facilitate data exchange and multi-library interactions.
  91. .. note:: This class doesn't manage the stream life-cycle, it is the user
  92. responsibility to keep the referenced stream alive while this class is
  93. being used.
  94. Args:
  95. stream_ptr(int): Integer representation of the `cudaStream_t` value.
  96. allocated externally.
  97. device(torch.device or int, optional): the device where the stream
  98. was originally allocated. if device is specified incorrectly,
  99. subsequent launches using this stream may fail.
  100. """
  101. def __new__(cls, stream_ptr, device=None, **kwargs):
  102. with torch.cuda.device(device):
  103. return super(ExternalStream, cls).__new__(cls, stream_ptr=stream_ptr, **kwargs)
  104. class Event(torch._C._CudaEventBase):
  105. r"""Wrapper around a CUDA event.
  106. CUDA events are synchronization markers that can be used to monitor the
  107. device's progress, to accurately measure timing, and to synchronize CUDA
  108. streams.
  109. The underlying CUDA events are lazily initialized when the event is first
  110. recorded or exported to another process. After creation, only streams on the
  111. same device may record the event. However, streams on any device can wait on
  112. the event.
  113. Args:
  114. enable_timing (bool, optional): indicates if the event should measure time
  115. (default: ``False``)
  116. blocking (bool, optional): if ``True``, :meth:`wait` will be blocking (default: ``False``)
  117. interprocess (bool): if ``True``, the event can be shared between processes
  118. (default: ``False``)
  119. .. _CUDA Event Documentation:
  120. https://docs.nvidia.com/cuda/cuda-runtime-api/group__CUDART__EVENT.html
  121. """
  122. def __new__(cls, enable_timing=False, blocking=False, interprocess=False):
  123. return super(Event, cls).__new__(
  124. cls,
  125. enable_timing=enable_timing, blocking=blocking, interprocess=interprocess)
  126. @classmethod
  127. def from_ipc_handle(cls, device, handle):
  128. r"""Reconstruct an event from an IPC handle on the given device."""
  129. return super(Event, cls).from_ipc_handle(device, handle)
  130. def record(self, stream=None):
  131. r"""Records the event in a given stream.
  132. Uses ``torch.cuda.current_stream()`` if no stream is specified. The
  133. stream's device must match the event's device."""
  134. if stream is None:
  135. stream = torch.cuda.current_stream()
  136. super().record(stream)
  137. def wait(self, stream=None):
  138. r"""Makes all future work submitted to the given stream wait for this
  139. event.
  140. Use ``torch.cuda.current_stream()`` if no stream is specified.
  141. .. note:: This is a wrapper around ``cudaStreamWaitEvent()``: see
  142. `CUDA Event documentation`_ for more info.
  143. """
  144. if stream is None:
  145. stream = torch.cuda.current_stream()
  146. super().wait(stream)
  147. def query(self):
  148. r"""Checks if all work currently captured by event has completed.
  149. Returns:
  150. A boolean indicating if all work currently captured by event has
  151. completed.
  152. """
  153. return super().query()
  154. def elapsed_time(self, end_event):
  155. r"""Returns the time elapsed in milliseconds after the event was
  156. recorded and before the end_event was recorded.
  157. """
  158. return super().elapsed_time(end_event)
  159. def synchronize(self):
  160. r"""Waits for the event to complete.
  161. Waits until the completion of all work currently captured in this event.
  162. This prevents the CPU thread from proceeding until the event completes.
  163. .. note:: This is a wrapper around ``cudaEventSynchronize()``: see
  164. `CUDA Event documentation`_ for more info.
  165. """
  166. super().synchronize()
  167. def ipc_handle(self):
  168. r"""Returns an IPC handle of this event. If not recorded yet, the event
  169. will use the current device. """
  170. return super().ipc_handle()
  171. @property
  172. def _as_parameter_(self):
  173. return ctypes.c_void_p(self.cuda_event)
  174. def __repr__(self):
  175. if self.cuda_event:
  176. return '<torch.cuda.Event {0:#x}>'.format(self._as_parameter_.value)
  177. else:
  178. return '<torch.cuda.Event uninitialized>'