class_docstr: All-reduce cross device ops using collective ops.
  
  In the between-graph replicated training, it will still do all-reduces across
  all workers and then put results on the right destinations.
class_docstr_tok: ["All-reduce","cross","device","ops","using","collective","ops",".","In","the","between-graph","replicated","training",",","it","will","still","do","all-reduces","across","all","workers","and","then","put","results","on","the","right","destinations","."]
code: def reduce_implementation(
      self, reduce_op, per_replica_value, destinations
  ):
      all_reduced = self._batch_all_reduce(
          reduce_op, [per_replica_value]
      )[0]
      device_map, logical_device = get_device_map_from(destinations)
      devices = device_map.logical_to_actual_devices(logical_device)
  
      if (
          isinstance(all_reduced, value_lib.Mirrored)
          and all_reduced.device_map is device_map
          and all_reduced.logical_device == logical_device
      ):
          return all_reduced
  
      # Convert `all_reduced` to a `Mirrored` object, as a simple and uniform
      # utility to access component for a particular device.
      if not isinstance(all_reduced, value_lib.Mirrored):
          all_reduced = value_lib.Mirrored(
              value_lib.SingleDeviceMap(all_reduced.device),
              [all_reduced],
          )
  
      index = []
      with ops.control_dependencies(all_reduced.values):
          for d in devices:
              with ops.device(d):
                  if d in all_reduced.devices:
                      index.append(
                          array_ops.identity(all_reduced.get(d))
                      )
                  else:
                      # TODO(josh11b): Once we add support for model parallelism, get the
                      # copy from the corresponding replica instead of the primary.
                      index.append(
                          array_ops.identity(all_reduced.primary)
                      )
  
      return value_lib.regroup(
          device_map, index, wrap_class=value_lib.Mirrored
      )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  import enum
  import six
  from tensorflow.python.client import device_lib
  from tensorflow.python.distribute import cross_device_utils
  from tensorflow.python.distribute import device_util
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.framework import kernels
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor_util
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import resource_variable_ops
  from tensorflow.python.platform import tf_logging
  from tensorflow.python.util.tf_export import tf_export
  from tensorflow.tools.docs import doc_controls
  import tensorflow
  from tensorflow.python.distribute.cross_device_ops import CollectiveAllReduce
  import tensorflow.python.distribute.cross_device_ops
  from byteps.tensorflow.ops import broadcast, _push_pull
  from byteps.tensorflow import local_rank, size
label_desc: Use image node_id 3 for calling the MyCollectiveAllReduce obj's underlying member method code with example usage: obj.reduce_implementation(reduce_op, per_replica_value, destinations) and returns: value_lib, all_reduced