replicate.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203
  1. from collections import OrderedDict
  2. from collections.abc import Iterator, Sequence
  3. from typing import cast, Optional, TYPE_CHECKING, TypeVar, Union
  4. from typing_extensions import TypeIs
  5. import torch
  6. from torch._utils import _get_device_index
  7. from torch.nn.modules import Module
  8. from torch.nn.parallel import comm
  9. if TYPE_CHECKING:
  10. from torch._C import ScriptMethod
  11. from torch.jit import ScriptModule
  12. from torch.jit._state import EnabledProxy
  13. __all__ = ["replicate"]
  14. def _is_script_module(module: Module) -> TypeIs["ScriptModule"]:
  15. import torch.jit
  16. return isinstance(module, torch.jit.ScriptModule)
  17. def _is_script_method(module: object) -> TypeIs["ScriptMethod"]:
  18. import torch.jit
  19. return isinstance(module, torch._C.ScriptMethod)
  20. def _init_script_module() -> "ScriptModule":
  21. import torch.jit
  22. return torch.jit.ScriptModule()
  23. def _is_jit_enabled() -> "EnabledProxy":
  24. import torch.jit._state
  25. return torch.jit._state._enabled
  26. # Check if we can safely replicate the module.
  27. # there are two types of module:
  28. # 1. python modules
  29. # 2. ScriptModule
  30. #
  31. # currently a module cannot be replicated properly if the descendants of
  32. # any ScriptModule contains python module (type 1 above)
  33. def _replicatable_module(module: Module, memo: Optional[set[Module]] = None) -> bool:
  34. # module.modules() contains module itself as the first element
  35. def descendant_modules(module: Module) -> Iterator[Module]:
  36. gen = module.modules()
  37. next(gen)
  38. return gen
  39. if not _is_jit_enabled():
  40. return True
  41. if memo is None:
  42. memo = set()
  43. # memoize visited modules
  44. memo.add(module)
  45. if _is_script_module(module):
  46. memo.update(descendant_modules(module))
  47. return all(
  48. _is_script_module(descendant) for descendant in descendant_modules(module)
  49. )
  50. for child in module.children():
  51. # since any unreplicatable module will cause the check to return
  52. # False early, visited modules here can be safely ignored.
  53. if child in memo:
  54. continue
  55. if not _replicatable_module(child, memo):
  56. return False
  57. return True
  58. def _broadcast_coalesced_reshape(
  59. tensors: Sequence[torch.Tensor],
  60. devices: Sequence[Union[int, torch.device]],
  61. detach: bool = False,
  62. ) -> list[list[torch.Tensor]]:
  63. from torch.nn.parallel._functions import Broadcast
  64. if detach:
  65. return comm.broadcast_coalesced(tensors, devices)
  66. else:
  67. # Use the autograd function to broadcast if not detach
  68. if len(tensors) > 0:
  69. tensor_copies = Broadcast.apply(devices, *tensors)
  70. return [
  71. tensor_copies[i : i + len(tensors)]
  72. for i in range(0, len(tensor_copies), len(tensors))
  73. ]
  74. else:
  75. return []
  76. T = TypeVar("T", bound=Module)
  77. def replicate(
  78. network: T,
  79. devices: Sequence[Union[int, torch.device]],
  80. detach: bool = False,
  81. ) -> list[T]:
  82. if not _replicatable_module(network):
  83. raise RuntimeError(
  84. "Cannot replicate network where python modules are children of ScriptModule"
  85. )
  86. if not devices:
  87. return []
  88. devices = [_get_device_index(x, True) for x in devices]
  89. num_replicas = len(devices)
  90. params = list(network.parameters())
  91. param_indices = {param: idx for idx, param in enumerate(params)}
  92. param_copies = _broadcast_coalesced_reshape(params, devices, detach)
  93. buffers = list(network.buffers())
  94. buffers_rg: list[torch.Tensor] = []
  95. buffers_not_rg: list[torch.Tensor] = []
  96. for buf in buffers:
  97. if buf.requires_grad and not detach:
  98. buffers_rg.append(buf)
  99. else:
  100. buffers_not_rg.append(buf)
  101. buffer_indices_rg = {buf: idx for idx, buf in enumerate(buffers_rg)}
  102. buffer_indices_not_rg = {buf: idx for idx, buf in enumerate(buffers_not_rg)}
  103. buffer_copies_rg = _broadcast_coalesced_reshape(buffers_rg, devices, detach=detach)
  104. buffer_copies_not_rg = _broadcast_coalesced_reshape(
  105. buffers_not_rg, devices, detach=True
  106. )
  107. modules = list(network.modules())
  108. module_copies: list[list[Module]] = [[] for _ in devices]
  109. module_indices: dict[Module, int] = {}
  110. for i, module in enumerate(modules):
  111. module_indices[module] = i
  112. for j in range(num_replicas):
  113. replica = module._replicate_for_data_parallel()
  114. # This is a temporary fix for DDP. DDP needs to access the
  115. # replicated model parameters. It used to do so through
  116. # `mode.parameters()`. The fix added in #33907 for DP stops the
  117. # `parameters()` API from exposing the replicated parameters.
  118. # Hence, we add a `_former_parameters` dict here to support DDP.
  119. replica._former_parameters = OrderedDict()
  120. module_copies[j].append(replica)
  121. for i, module in enumerate(modules):
  122. for key, child in module._modules.items():
  123. if child is None:
  124. for j in range(num_replicas):
  125. replica = module_copies[j][i]
  126. replica._modules[key] = None
  127. else:
  128. module_idx = module_indices[child]
  129. for j in range(num_replicas):
  130. replica = module_copies[j][i]
  131. setattr(replica, key, module_copies[j][module_idx])
  132. for key, param in module._parameters.items():
  133. if param is None:
  134. for j in range(num_replicas):
  135. replica = module_copies[j][i]
  136. replica._parameters[key] = None
  137. else:
  138. param_idx = param_indices[param]
  139. for j in range(num_replicas):
  140. replica = module_copies[j][i]
  141. param_copy = param_copies[j][param_idx]
  142. # parameters in replicas are no longer leaves,
  143. # so setattr them as non-parameter attributes
  144. setattr(replica, key, param_copy)
  145. # expose the parameter for DDP
  146. replica._former_parameters[key] = param_copy # type: ignore[operator, index]
  147. for key, buf in module._buffers.items(): # type: ignore[assignment]
  148. if buf is None:
  149. for j in range(num_replicas):
  150. replica = module_copies[j][i]
  151. replica._buffers[key] = None
  152. else:
  153. if buf.requires_grad and not detach:
  154. buffer_copies = buffer_copies_rg
  155. buffer_idx = buffer_indices_rg[buf]
  156. else:
  157. buffer_copies = buffer_copies_not_rg
  158. buffer_idx = buffer_indices_not_rg[buf]
  159. for j in range(num_replicas):
  160. replica = module_copies[j][i]
  161. setattr(replica, key, buffer_copies[j][buffer_idx])
  162. return [cast(T, module_copies[j][0]) for j in range(num_replicas)]