code: def testAllReduceSparse(
      self, num_processes, required_gpus, implementation, reduce_op
  ):
      if (
          required_gpus == 0
          and implementation == CommunicationImplementation.NCCL
      ):
          self.skipTest("Skip CPU + NCCL combination")
      if (
          num_processes != required_gpus
          and implementation == CommunicationImplementation.NCCL
      ):
          self.skipTest(
              "Skip NCCL combination with mismatched process and GPU "
              "count. NCCL requires physical GPUs for every process."
          )
      if (
          num_processes != required_gpus
          and implementation == CommunicationImplementation.AUTO
      ):
          self.skipTest(
              "Skip potential NCCL combination (AUTO) with mismatched "
              "process and GPU count. NCCL requires physical GPUs for "
              "every process."
          )
  
      def replica_fn():
          collective, devices, _ = self.make_collective(
              num_processes, required_gpus
          )
          options = collective_util.Options(
              implementation=implementation
          )
          group_size = num_processes * (required_gpus or 1)
  
          @def_function.function
          def collective_all_reduce():
              results = []
              for replica_id, device in enumerate(devices):
                  with ops.device(device):
                      value = IndexedSlices(
                          values=array_ops.identity([[1.0]]),
                          indices=array_ops.identity([0]),
                          dense_shape=array_ops.identity([5, 1]),
                      )
                      results.append(
                          collective._all_reduce(
                              reduce_op, value, replica_id, options
                          )
                      )
              return results
  
          got = collective_all_reduce()
          if reduce_op == ReduceOp.SUM:
              expect = [
                  IndexedSlices([[1.0 * group_size]], [0], [5, 1])
              ] * len(devices)
          elif reduce_op == ReduceOp.MEAN:
              expect = [IndexedSlices([[1.0]], [0], [5, 1])] * len(
                  devices
              )
          self.assertAllClose(
              nest.map_structure(ops.convert_to_tensor, got),
              nest.map_structure(ops.convert_to_tensor, expect),
          )
  
          @def_function.function
          def collective_batch_all_reduce():
              results = []
              for replica_id, device in enumerate(devices):
                  with ops.device(device):
                      value = (
                          IndexedSlices(
                              array_ops.identity([[1.0]]),
                              array_ops.identity([0]),
                              array_ops.identity([5, 1]),
                          ),
                          IndexedSlices(
                              array_ops.identity([[3.0]]),
                              array_ops.identity([2]),
                              array_ops.identity([5, 1]),
                          ),
                      )
                      results.append(
                          collective._all_reduce(
                              reduce_op, value, replica_id, options
                          )
                      )
              return results
  
          got = collective_batch_all_reduce()
          if reduce_op == ReduceOp.SUM:
              expect = [
                  (
                      IndexedSlices([[1.0 * group_size]], [0], [5, 1]),
                      IndexedSlices([[3.0 * group_size]], [2], [5, 1]),
                  )
              ] * len(devices)
          elif reduce_op == ReduceOp.MEAN:
              expect = [
                  (
                      IndexedSlices([[1.0]], [0], [5, 1]),
                      IndexedSlices([[3.0]], [2], [5, 1]),
                  )
              ] * len(devices)
          self.assertAllClose(
              nest.map_structure(ops.convert_to_tensor, got),
              nest.map_structure(ops.convert_to_tensor, expect),
          )
  
      get_global_mpr(num_processes).run(replica_fn)
  
imports: import collections
  import os
  import threading
  import time
  from absl.testing import parameterized
  from tensorflow.core.protobuf import config_pb2
  from tensorflow.core.protobuf import tensorflow_server_pb2
  from tensorflow.python.distribute import cluster_resolver
  from tensorflow.python.distribute import collective_util
  from tensorflow.python.distribute import combinations
  from tensorflow.python.distribute import cross_device_ops
  from tensorflow.python.distribute import cross_device_utils
  from tensorflow.python.distribute import device_util
  from tensorflow.python.distribute import multi_process_runner
  from tensorflow.python.distribute import multi_worker_test_base
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import test_util
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.eager import def_function
  from tensorflow.python.eager import test
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import dtypes
  from tensorflow.python.framework import errors
  from tensorflow.python.framework import indexed_slices
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import collective_ops
  from tensorflow.python.ops import cond
  from tensorflow.python.ops import math_ops
  from tensorflow.python.util import nest
label_desc: Use image node_id 13 for calling the CollectiveOpsTest obj's underlying member method code with example usage: obj.testAllReduceSparse(num_processes, required_gpus, implementation, reduce_op) and returns: results, results