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 __init__(
      self,
      module,
      message_size=10000000,
      delay_allreduce=False,
      shared_param=None,
      allreduce_trigger_params=None,
      retain_allreduce_buffers=False,
      allreduce_always_fp32=False,
      num_allreduce_streams=1,
      allreduce_communicators=None,
      gradient_average=True,
      gradient_predivide_factor=1.0,
      gradient_average_split_factor=None,
      prof=False,
  ):
      super(DistributedDataParallel, self).__init__()
      from apex import deprecated_warning
  
      deprecated_warning(
          "apex.parallel.DistributedDataParallel is deprecated and will be removed by the end of February 2023."
      )
  
      # Backward/forward compatibility around
      # https://github.com/pytorch/pytorch/commit/540ef9b1fc5506369a48491af8a285a686689b36 and
      # https://github.com/pytorch/pytorch/commit/044d00516ccd6572c0d6ab6d54587155b02a3b86
      if hasattr(dist, "get_backend"):
          self._backend = dist.get_backend()
          if hasattr(dist, "DistBackend"):
              self.backend_enum_holder = dist.DistBackend
          else:
              self.backend_enum_holder = dist.Backend
      else:
          self._backend = dist._backend
          self.backend_enum_holder = dist.dist_backend
  
      self.warn_on_half = (
          True
          if self._backend == self.backend_enum_holder.GLOO
          else False
      )
  
      self.prof = prof
  
      self.allreduce_different_streams = num_allreduce_streams > 1
      self.num_allreduce_streams = num_allreduce_streams
      self.allreduce_communicators = allreduce_communicators
      if self.allreduce_communicators:
          assert (
              len(allreduce_communicators[0]) == num_allreduce_streams
          )
          assert len(allreduce_communicators[0]) == len(
              allreduce_communicators[1]
          )
          assert self.allreduce_different_streams
  
      if self.allreduce_different_streams and delay_allreduce:
          raise ValueError(
              "self.allreduce_different_streams may only be used if delay_allreduce=False."
          )
  
      if shared_param is not None:
          raise ValueError(
              "shared_param is no longer supported as an option.  It was misleadingly named from the start.  It turns out overlapping communication with computation should work fine with shared parameters.  If you still wish to delay communication to the end of the backward pass, use delay_allreduce=True|False instead."
          )
  
      self.world_size = float(dist.get_world_size())
  
      self.retain_allreduce_buffers = retain_allreduce_buffers
      self.allreduce_always_fp32 = allreduce_always_fp32
      self.gradient_average = gradient_average
      self.gradient_predivide_factor = gradient_predivide_factor
  
      self.custom_allreduce_triggers = False
      if allreduce_trigger_params is not None:
          if delay_allreduce:
              raise ValueError(
                  "Setting allreduce_trigger_params is only valid if delay_allreduce=False."
              )
          self.custom_allreduce_triggers = True
          self.allreduce_trigger_params = set(
              [id(param) for param in allreduce_trigger_params]
          )
  
      self.delay_allreduce = delay_allreduce
      self.message_size = message_size
  
      self.main_stream = torch.cuda.current_stream()
  
      self.bucket_streams = []
      self.bucket_events = []
  
      self.module = module
  
      self._disable_allreduce = False
  
      if self._backend == self.backend_enum_holder.NCCL:
          for param in self.module.parameters():
              assert (
                  param.is_cuda
              ), "NCCL backend only supports model parameters to be on GPU."
  
      self.active_params = []
  
      self.param_type_to_tmp_i = {
          "torch.cuda.HalfTensor": 0,
          "torch.cuda.FloatTensor": 1,
          "torch.cuda.DoubleTensor": 2,
      }
  
      if multi_tensor_applier.available:
          # TODO:  I really need to centralize the C++ backed imports
          import amp_C
  
          self.multi_tensor_scale = amp_C.multi_tensor_scale
          self._overflow_buf = torch.cuda.IntTensor([0])
  
      self.create_hooks()
  
      flat_dist_call(
          [param.data for param in self.module.parameters()],
          dist.broadcast,
          (0,),
      )
  
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 1 to create a new DistributedDataParallel object from inherited base classes: Module with example: obj = DistributedDataParallel(module, message_size, delay_allreduce, shared_param, allreduce_trigger_params, retain_allreduce_buffers, allreduce_always_fp32, num_allreduce_streams, allreduce_communicators, gradient_average, gradient_predivide_factor, gradient_average_split_factor, prof)