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 allreduce_bucket(self, bucket, bucket_idx, force_default_stream):
      tensor = flatten(bucket)
  
      if force_default_stream:
          bucket_stream = self.main_stream
      else:
          bucket_stream = self._stream_this_bucket(bucket_idx)
          bucket_event = self._event_this_bucket(bucket_idx)
          torch.cuda.current_stream().record_event(bucket_event)
          bucket_stream.wait_event(bucket_event)
  
      with torch.cuda.stream(bucket_stream):
          # self.main_stream.wait_stream(torch.cuda.current_stream())
          # torch.cuda.synchronize()
  
          tensor_to_allreduce = tensor
  
          if self.allreduce_always_fp32:
              tensor_to_allreduce = tensor.float()
  
          if self.gradient_predivide_factor != 1.0:
              tensor_to_allreduce.mul_(
                  1.0 / self.gradient_predivide_factor
              )
  
          if (
              self.allreduce_different_streams
              and not force_default_stream
          ):
              dist.all_reduce(
                  tensor_to_allreduce,
                  group=self.bucket_pgs[
                      bucket_idx % self.num_allreduce_streams
                  ],
              )
          else:
              dist.all_reduce(tensor_to_allreduce)
  
          if self.gradient_average:
              tensor_to_allreduce.mul_(
                  self.gradient_predivide_factor / self.world_size
              )
  
          if (
              self.allreduce_always_fp32
              and tensor is not tensor_to_allreduce
          ):
              tensor.copy_(tensor_to_allreduce)
  
          if not self.retain_allreduce_buffers:
              if multi_tensor_applier.available:
                  multi_tensor_applier(
                      self.multi_tensor_scale,
                      self._overflow_buf,
                      [unflatten(tensor, bucket), bucket],
                      1.0,
                  )
              else:
                  for buf, synced in zip(
                      bucket, unflatten(tensor, bucket)
                  ):
                      buf.copy_(synced)
  
          # I think we actually do need this here.  After allreduce_bucket returns, tensor will
          # eventually go out of scope and die, at which point it could otherwise be freed for
          # further reuse by the main stream while the allreduce/div/unflatten are underway in bucket_stream.
          tensor.record_stream(bucket_stream)
  
      return tensor
  
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 10 for calling the DistributedDataParallel obj's underlying member method code with example usage: obj.allreduce_bucket(bucket, bucket_idx, force_default_stream) and returns: tensor