_functions.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315
  1. # mypy: allow-untyped-defs
  2. import torch
  3. import torch.distributed as dist
  4. from torch.autograd.function import Function
  5. class SyncBatchNorm(Function):
  6. @staticmethod
  7. def forward(
  8. self,
  9. input,
  10. weight,
  11. bias,
  12. running_mean,
  13. running_var,
  14. eps,
  15. momentum,
  16. process_group,
  17. world_size,
  18. ):
  19. if not (
  20. input.is_contiguous(memory_format=torch.channels_last)
  21. or input.is_contiguous(memory_format=torch.channels_last_3d)
  22. ):
  23. input = input.contiguous()
  24. if weight is not None:
  25. weight = weight.contiguous()
  26. size = int(input.numel() // input.size(1))
  27. if size == 1 and world_size < 2:
  28. raise ValueError(
  29. f"Expected more than 1 value per channel when training, got input size {size}"
  30. )
  31. num_channels = input.shape[1]
  32. if input.numel() > 0:
  33. # calculate mean/invstd for input.
  34. mean, invstd = torch.batch_norm_stats(input, eps)
  35. count = torch.full(
  36. (1,),
  37. input.numel() // input.size(1),
  38. dtype=mean.dtype,
  39. device=mean.device,
  40. )
  41. # C, C, 1 -> (2C + 1)
  42. combined = torch.cat([mean, invstd, count], dim=0)
  43. else:
  44. # for empty input, set stats and the count to zero. The stats with
  45. # zero count will be filtered out later when computing global mean
  46. # & invstd, but they still needs to participate the all_gather
  47. # collective communication to unblock other peer processes.
  48. combined = torch.zeros(
  49. 2 * num_channels + 1, dtype=input.dtype, device=input.device
  50. )
  51. # Use allgather instead of allreduce because count could be different across
  52. # ranks, simple all reduce op can not give correct results.
  53. # batch_norm_gather_stats_with_counts calculates global mean & invstd based on
  54. # all gathered mean, invstd and count.
  55. # for nccl backend, use the optimized version of all gather.
  56. # The Gloo backend does not support `all_gather_into_tensor`.
  57. if process_group._get_backend_name() != "gloo":
  58. # world_size * (2C + 1)
  59. combined_size = combined.numel()
  60. combined_flat = torch.empty(
  61. 1,
  62. combined_size * world_size,
  63. dtype=combined.dtype,
  64. device=combined.device,
  65. )
  66. dist.all_gather_into_tensor(
  67. combined_flat, combined, process_group, async_op=False
  68. )
  69. combined = torch.reshape(combined_flat, (world_size, combined_size))
  70. # world_size * (2C + 1) -> world_size * C, world_size * C, world_size * 1
  71. mean_all, invstd_all, count_all = torch.split(combined, num_channels, dim=1)
  72. else:
  73. # world_size * (2C + 1)
  74. combined_list = [torch.empty_like(combined) for _ in range(world_size)]
  75. dist.all_gather(combined_list, combined, process_group, async_op=False)
  76. combined = torch.stack(combined_list, dim=0)
  77. # world_size * (2C + 1) -> world_size * C, world_size * C, world_size * 1
  78. mean_all, invstd_all, count_all = torch.split(combined, num_channels, dim=1)
  79. if not (torch.cuda.is_available() and torch.cuda.is_current_stream_capturing()):
  80. # The lines below force a synchronization between CUDA and CPU, because
  81. # the shape of the result count_all depends on the values in mask tensor.
  82. # Such synchronizations break CUDA Graph capturing.
  83. # See https://github.com/pytorch/pytorch/issues/78549
  84. # FIXME: https://github.com/pytorch/pytorch/issues/78656 describes
  85. # a better longer-term solution.
  86. # remove stats from empty inputs
  87. mask = count_all.squeeze(-1) >= 1
  88. count_all = count_all[mask]
  89. mean_all = mean_all[mask]
  90. invstd_all = invstd_all[mask]
  91. # calculate global mean & invstd
  92. counts = count_all.view(-1)
  93. if running_mean is not None and counts.dtype != running_mean.dtype:
  94. counts = counts.to(running_mean.dtype)
  95. mean, invstd = torch.batch_norm_gather_stats_with_counts(
  96. input,
  97. mean_all,
  98. invstd_all,
  99. running_mean,
  100. running_var,
  101. momentum,
  102. eps,
  103. counts,
  104. )
  105. self.save_for_backward(input, weight, mean, invstd, count_all.to(torch.int32))
  106. self.process_group = process_group
  107. # apply element-wise normalization
  108. if input.numel() > 0:
  109. return torch.batch_norm_elemt(input, weight, bias, mean, invstd, eps)
  110. else:
  111. return torch.empty_like(input)
  112. @staticmethod
  113. def backward(self, grad_output):
  114. if not (
  115. grad_output.is_contiguous(memory_format=torch.channels_last)
  116. or grad_output.is_contiguous(memory_format=torch.channels_last_3d)
  117. ):
  118. grad_output = grad_output.contiguous()
  119. saved_input, weight, mean, invstd, count_tensor = self.saved_tensors
  120. grad_input = grad_weight = grad_bias = None
  121. process_group = self.process_group
  122. if saved_input.numel() > 0:
  123. # calculate local stats as well as grad_weight / grad_bias
  124. (
  125. sum_dy,
  126. sum_dy_xmu,
  127. grad_weight,
  128. grad_bias,
  129. ) = torch.batch_norm_backward_reduce(
  130. grad_output,
  131. saved_input,
  132. mean,
  133. invstd,
  134. weight,
  135. self.needs_input_grad[0],
  136. self.needs_input_grad[1],
  137. self.needs_input_grad[2],
  138. )
  139. if self.needs_input_grad[0]:
  140. # synchronizing stats used to calculate input gradient.
  141. num_channels = sum_dy.shape[0]
  142. combined = torch.cat([sum_dy, sum_dy_xmu], dim=0)
  143. torch.distributed.all_reduce(
  144. combined,
  145. torch.distributed.ReduceOp.SUM,
  146. process_group,
  147. async_op=False,
  148. )
  149. sum_dy, sum_dy_xmu = torch.split(combined, num_channels)
  150. # backward pass for gradient calculation
  151. if weight is not None and weight.dtype != mean.dtype:
  152. weight = weight.to(mean.dtype)
  153. grad_input = torch.batch_norm_backward_elemt(
  154. grad_output,
  155. saved_input,
  156. mean,
  157. invstd,
  158. weight,
  159. sum_dy,
  160. sum_dy_xmu,
  161. count_tensor,
  162. )
  163. # synchronizing of grad_weight / grad_bias is not needed as distributed
  164. # training would handle all reduce.
  165. if weight is None or not self.needs_input_grad[1]:
  166. grad_weight = None
  167. if weight is None or not self.needs_input_grad[2]:
  168. grad_bias = None
  169. else:
  170. # This process got an empty input tensor in the forward pass.
  171. # Although this process can directly set grad_input as an empty
  172. # tensor of zeros, it still needs to participate in the collective
  173. # communication to unblock its peers, as other peer processes might
  174. # have received non-empty inputs.
  175. num_channels = saved_input.shape[1]
  176. if self.needs_input_grad[0]:
  177. # launch all_reduce to unblock other peer processes
  178. combined = torch.zeros(
  179. 2 * num_channels, dtype=saved_input.dtype, device=saved_input.device
  180. )
  181. torch.distributed.all_reduce(
  182. combined,
  183. torch.distributed.ReduceOp.SUM,
  184. process_group,
  185. async_op=False,
  186. )
  187. # Leave grad_input, grad_weight and grad_bias as None, which will be
  188. # interpreted by the autograd engine as Tensors full of zeros.
  189. return grad_input, grad_weight, grad_bias, None, None, None, None, None, None
  190. class CrossMapLRN2d(Function):
  191. @staticmethod
  192. def forward(ctx, input, size, alpha=1e-4, beta=0.75, k=1):
  193. ctx.size = size
  194. ctx.alpha = alpha
  195. ctx.beta = beta
  196. ctx.k = k
  197. ctx.scale = None
  198. if input.dim() != 4:
  199. raise ValueError(
  200. f"CrossMapLRN2d: Expected input to be 4D, got {input.dim()}D instead."
  201. )
  202. ctx.scale = ctx.scale or input.new()
  203. output = input.new()
  204. channels = input.size(1)
  205. output.resize_as_(input)
  206. ctx.scale.resize_as_(input)
  207. # use output storage as temporary buffer
  208. input_square = output
  209. torch.pow(input, 2, out=input_square)
  210. pre_pad = int((ctx.size - 1) / 2 + 1)
  211. pre_pad_crop = min(pre_pad, channels)
  212. scale_first = ctx.scale.select(1, 0)
  213. scale_first.zero_()
  214. # compute first feature map normalization
  215. for c in range(pre_pad_crop):
  216. scale_first.add_(input_square.select(1, c))
  217. # reuse computations for next feature maps normalization
  218. # by adding the next feature map and removing the previous
  219. for c in range(1, channels):
  220. scale_previous = ctx.scale.select(1, c - 1)
  221. scale_current = ctx.scale.select(1, c)
  222. scale_current.copy_(scale_previous)
  223. if c < channels - pre_pad + 1:
  224. square_next = input_square.select(1, c + pre_pad - 1)
  225. scale_current.add_(square_next, alpha=1)
  226. if c > pre_pad:
  227. square_previous = input_square.select(1, c - pre_pad)
  228. scale_current.add_(square_previous, alpha=-1)
  229. ctx.scale.mul_(ctx.alpha / ctx.size).add_(ctx.k)
  230. torch.pow(ctx.scale, -ctx.beta, out=output)
  231. output.mul_(input)
  232. ctx.save_for_backward(input, output)
  233. return output
  234. @staticmethod
  235. def backward(ctx, grad_output):
  236. input, output = ctx.saved_tensors
  237. grad_input = grad_output.new()
  238. batch_size = input.size(0)
  239. channels = input.size(1)
  240. input_height = input.size(2)
  241. input_width = input.size(3)
  242. paddded_ratio = input.new(channels + ctx.size - 1, input_height, input_width)
  243. accum_ratio = input.new(input_height, input_width)
  244. cache_ratio_value = 2 * ctx.alpha * ctx.beta / ctx.size
  245. inversePrePad = int(ctx.size - (ctx.size - 1) / 2)
  246. grad_input.resize_as_(input)
  247. torch.pow(ctx.scale, -ctx.beta, out=grad_input).mul_(grad_output)
  248. paddded_ratio.zero_()
  249. padded_ratio_center = paddded_ratio.narrow(0, inversePrePad, channels)
  250. for n in range(batch_size):
  251. torch.mul(grad_output[n], output[n], out=padded_ratio_center)
  252. padded_ratio_center.div_(ctx.scale[n])
  253. torch.sum(
  254. paddded_ratio.narrow(0, 0, ctx.size - 1),
  255. 0,
  256. keepdim=False,
  257. out=accum_ratio,
  258. )
  259. for c in range(channels):
  260. accum_ratio.add_(paddded_ratio[c + ctx.size - 1])
  261. grad_input[n][c].addcmul_(
  262. input[n][c], accum_ratio, value=-cache_ratio_value
  263. )
  264. accum_ratio.add_(paddded_ratio[c], alpha=-1)
  265. return grad_input, None, None, None, None
  266. class BackwardHookFunction(torch.autograd.Function):
  267. @staticmethod
  268. def forward(ctx, *args):
  269. ctx.mark_non_differentiable(*[arg for arg in args if not arg.requires_grad])
  270. return args
  271. @staticmethod
  272. def backward(ctx, *args):
  273. return args