_functions.py 11 KB

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