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 forward(self, *inputs, **kwargs):
      result = self.module(*inputs, **kwargs)
  
      if self.prof:
          torch.cuda.nvtx.range_push("forward pass DDP logic")
  
      if not self._disable_allreduce:
          if not self.delay_allreduce:
              param_list = [
                  param
                  for param in self.module.parameters()
                  if param.requires_grad
              ]
  
              # Conditions under which to refresh self.record
              # Forward has the authority to set needs_refresh to True, but only allreduce_params
              # in backward has the authority to set needs_refresh to False.
              # Parentheses are not necessary for correct order of operations, but make the intent clearer.
              if (
                  (not self.active_params)
                  or (len(param_list) != len(self.active_params))
                  or any(
                      [
                          param1 is not param2
                          for param1, param2 in zip(
                              param_list, self.active_params
                          )
                      ]
                  )
              ):
                  self.needs_refresh = True
  
              if self.needs_refresh:
                  self.active_i_buckets = []
                  self.buckets = []
                  self.tmp_buckets = [
                      [],
                      [],
                      [],
                  ]  # [running half, float, double buckets]
                  self.tmp_numels = [0, 0, 0]
                  self.bucket_sizes = []
                  self.param_id_to_active_i = {
                      id(param): i for i, param in enumerate(param_list)
                  }
                  self.param_id_to_bucket = {}
                  self.bucket_pgs = []
                  self.bucket_streams = []
                  self.bucket_events = []
              else:
                  # self.buckets = [[None for _ in range(self.bucket_sizes[i])]
                  #                 for i in range(self.num_buckets)]
                  if not self.buckets:
                      self.buckets = [
                          [None for _ in range(self.bucket_sizes[i])]
                          for i in range(self.num_buckets)
                      ]
                  else:
                      assert (
                          len(self.buckets) == self.num_buckets
                      ), "len(buckets) = {}, expected {}".format(
                          len(self.buckets), self.num_buckets
                      )
                      for b, bucket in enumerate(self.buckets):
                          assert (
                              len(bucket) == self.bucket_sizes[b]
                          ), "len(buckets[{}]) = {}, expected {})".format(
                              b, len(buckets[b]), self.bucket_sizes[b]
                          )
                          for i in range(len(bucket)):
                              bucket[i] = None
  
                  if self.allreduce_communicators:
                      self.bucket_pgs = self.allreduce_communicators[0]
                      self.bucket_streams = (
                          self.allreduce_communicators[1]
                      )
                      self.bucket_events = [
                          torch.cuda.Event(
                              enable_timing=False, blocking=False
                          )
                          for _ in range(self.num_allreduce_streams)
                      ]
                  else:
                      if self.allreduce_different_streams:
                          if not self.bucket_pgs:
                              self.bucket_pgs = [
                                  dist.new_group()
                                  for _ in range(
                                      self.num_allreduce_streams
                                  )
                              ]
                              for i, bg in enumerate(self.bucket_pgs):
                                  print(
                                      "rank {} created group {} with backend {}".format(
                                          dist.get_rank(),
                                          i,
                                          dist.get_backend(bg),
                                      )
                                  )
                      if self.allreduce_different_streams:
                          if not self.bucket_streams:
                              self.bucket_streams = [
                                  torch.cuda.Stream()
                                  for _ in range(
                                      self.num_allreduce_streams
                                  )
                              ]
                              self.bucket_events = [
                                  torch.cuda.Event(
                                      enable_timing=False,
                                      blocking=False,
                                  )
                                  for _ in range(
                                      self.num_allreduce_streams
                                  )
                              ]
                      else:
                          if not self.bucket_streams:
                              self.bucket_streams = [
                                  torch.cuda.Stream()
                              ]
                              self.bucket_events = [
                                  torch.cuda.Event(
                                      enable_timing=False,
                                      blocking=False,
                                  )
                              ]
  
                  self.buckets_ready_size = [
                      0 for i in range(self.num_buckets)
                  ]
                  if self.retain_allreduce_buffers:
                      self.allreduce_buffers = [
                          None for _ in range(self.num_buckets)
                      ]
                  self.next_bucket = 0
                  self.ready_buckets_not_reduced = set()
  
              self.active_params = param_list
  
          self.callback_queued = False
  
      if self.prof:
          torch.cuda.nvtx.range_pop()
  
      return result
  
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 14 for calling the DistributedDataParallel obj's underlying member method code with example usage: obj.forward() and returns: result