class_docstr: :class:`apex.parallel.DistributedDataParallel` is a module wrapper that enables
  easy multiprocess distributed data parallel training, similar to ``torch.nn.parallel.DistributedDataParallel``.  Parameters are broadcast across participating processes on initialization, and gradients are
  allreduced and averaged over processes during ``backward()``.
  
  :class:`DistributedDataParallel` is optimized for use with NCCL.  It achieves high performance by
  overlapping communication with computation during ``backward()`` and bucketing smaller gradient
  transfers to reduce the total number of transfers required.
  
  :class:`DistributedDataParallel` is designed to work with the upstream launch utility script
  ``torch.distributed.launch`` with ``--nproc_per_node <= number of gpus per node``.
  When used with this launcher, :class:`DistributedDataParallel` assumes 1:1 mapping of processes to GPUs.
  It also assumes that your script calls ``torch.cuda.set_device(args.rank)`` before creating the model.
  
  https://github.com/NVIDIA/apex/tree/master/examples/simple/distributed shows detailed usage.
  https://github.com/NVIDIA/apex/tree/master/examples/imagenet shows another example
  that combines :class:`DistributedDataParallel` with mixed precision training.
  
  Args:
      module: Network definition to be run in multi-gpu/distributed mode.
      message_size (int, default=1e7): Minimum number of elements in a communication bucket.
      delay_allreduce (bool, default=False):  Delay all communication to the end of the backward pass.  This disables overlapping communication with computation.
      allreduce_trigger_params (list, optional, default=None):  If supplied, should contain a list of parameters drawn from the model.  Allreduces will be kicked off whenever one of these parameters receives its gradient (as opposed to when a bucket of size message_size is full).  At the end of backward(), a cleanup allreduce to catch any remaining gradients will also be performed automatically.  If allreduce_trigger_params is supplied, the message_size argument will be ignored.
      allreduce_always_fp32 (bool, default=False):  Convert any FP16 gradients to FP32 before allreducing.  This can improve stability for widely scaled-out runs.
      gradient_average (bool, default=True):  Option to toggle whether or not DDP averages the allreduced gradients over processes.  For proper scaling, the default value of True is recommended.
      gradient_predivide_factor (float, default=1.0):  Allows perfoming the average of gradients over processes partially before and partially after the allreduce.  Before allreduce:  ``grads.mul_(1.0/gradient_predivide_factor)``.  After allreduce:  ``grads.mul_(gradient_predivide_factor/world size)``.  This can reduce the stress on the dynamic range of FP16 allreduces for widely scaled-out runs.
  
  .. warning::
      If ``gradient_average=False``, the pre-allreduce division (``grads.mul_(1.0/gradient_predivide_factor)``) will still be applied, but the post-allreduce gradient averaging (``grads.mul_(gradient_predivide_factor/world size)``) will be omitted.
code: def create_hooks(self):
      # Fallback hook that's only called at the end of backward.
      # Used if you deliberately want to delay allreduces to the end, or to refresh the
      # bucket structure that will be used to overlap communication with computation in later
      # iterations.
      def allreduce_params():
          # Bucket record refresh
          if not self.delay_allreduce:
              if self.needs_refresh:
                  self.sync_bucket_structure()
  
                  self.needs_refresh = False
  
          self.allreduce_fallback()
  
      def overlapping_backward_epilogue():
          for stream, event in zip(
              self.bucket_streams, self.bucket_events
          ):
              stream.record_event(event)
              torch.cuda.current_stream().wait_event(event)
  
          # Sanity checks that all the buckets were kicked off
          if self.next_bucket != self.num_buckets:
              raise RuntimeError(
                  "In epilogue, next_bucket ({}) != num_buckets ({}).  ".format(
                      self.next_bucket, self.num_buckets
                  ),
                  "This probably indicates some buckets were not allreduced.",
              )
  
          for actual, expected in zip(
              self.buckets_ready_size, self.bucket_sizes
          ):
              if actual != expected:
                  raise RuntimeError(
                      "Some param buckets were not allreduced."
                  )
  
      self.grad_accs = []
      for param in self.module.parameters():
          if param.requires_grad:
  
              def wrapper(param):
                  param_tmp = param.expand_as(param)
                  grad_acc = param_tmp.grad_fn.next_functions[0][0]
  
                  def allreduce_hook(*unused):
                      if self.prof:
                          torch.cuda.nvtx.range_push("allreduce_hook")
  
                      if not self._disable_allreduce:
                          if self.delay_allreduce or self.needs_refresh:
                              # TODO:  How do we want to handle multiple backward passes between
                              # each forward, e.g., backward passes with retain_graph=True?
                              # needs_refresh and callback_queued are both vulnerable states.
                              if (
                                  not self.delay_allreduce
                                  and self.needs_refresh
                              ):
                                  # Use the backward pass to build the bucket structure on the fly.
                                  active_i = self.param_id_to_active_i[
                                      id(param)
                                  ]
  
                                  # Float, half, and double tensors are grouped into buckets separately.
                                  current_type = (
                                      self.param_type_to_tmp_i[
                                          param.type()
                                      ]
                                  )
  
                                  self.tmp_buckets[current_type].append(
                                      active_i
                                  )
  
                                  ship_tmp_bucket = False
                                  if self.custom_allreduce_triggers:
                                      if (
                                          id(param)
                                          in self.allreduce_trigger_params
                                      ):
                                          ship_tmp_bucket = True
                                  else:
                                      self.tmp_numels[
                                          current_type
                                      ] += param.numel()
                                      if (
                                          self.tmp_numels[current_type]
                                          >= self.message_size
                                      ):
                                          ship_tmp_bucket = True
  
                                  # To consider:  If custom_allreduce_triggers are in use, ship all
                                  # tmp_buckets, not just tmp_buckets[current_type].
                                  if ship_tmp_bucket:
                                      self.active_i_buckets.append(
                                          self.tmp_buckets[current_type]
                                      )
                                      self.tmp_buckets[
                                          current_type
                                      ] = []
                                      self.tmp_numels[current_type] = 0
  
                              if not self.callback_queued:
                                  Variable._execution_engine.queue_callback(
                                      allreduce_params
                                  )
                                  self.callback_queued = True
                          else:
                              if not self.callback_queued:
                                  Variable._execution_engine.queue_callback(
                                      overlapping_backward_epilogue
                                  )
                                  self.callback_queued = True
  
                              self.comm_ready_buckets(param)
  
                      if self.prof:
                          torch.cuda.nvtx.range_pop()
  
                  grad_acc.register_hook(allreduce_hook)
                  self.grad_accs.append(grad_acc)
  
              wrapper(param)
  
imports: from collections import OrderedDict
  import copy
  import importlib
  from itertools import chain
  import torch
  import torch.distributed
  from torch.nn.modules import Module
  from torch.autograd import Variable
  from ..multi_tensor_apply import multi_tensor_applier
label_desc: Use image node_id 7 for calling the DistributedDataParallel obj's underlying member method code with example usage: obj.create_hooks() without return types