code: def _test_reduce_indexed_slices(
      self,
      task_type,
      task_id,
      num_gpus,
      communication,
      batch_reduce,
      variable_length,
      local_mode=False,
  ):
      (
          collective_all_reduce,
          devices,
          master_target,
      ) = self._get_test_objects(
          task_type,
          task_id,
          num_gpus,
          communication=communication,
          local_mode=local_mode,
      )
      if local_mode:
          num_workers = 1
          worker_device = None
      else:
          num_workers = len(self._cluster_spec.get("chief", [])) + len(
              self._cluster_spec.get("worker", [])
          )
          worker_device = "/job:%s/task:%d" % (task_type, task_id)
          config = config_pb2.ConfigProto()
          # Disable coordination service for all tasks except task 0 to avoid
          # parallel init of the service singleton.
          if task_id != 0:
              config.experimental.coordination_config.service_type = ""
      with ops.Graph().as_default(), ops.device(
          worker_device
      ), self.cached_session(
          target=master_target, config=config
      ) as sess:
          per_replica = self._get_indexed_slices(
              devices,
              (task_id or 0) * max(num_gpus, 1),
              variable_length,
          )
  
          if batch_reduce:
              result = collective_all_reduce.batch_reduce(
                  reduce_util.ReduceOp.SUM, [(per_replica, per_replica)]
              )[0]
          else:
              result = collective_all_reduce.reduce(
                  reduce_util.ReduceOp.SUM, per_replica, per_replica
              )
          if num_gpus > 1:
              self.assertIsInstance(result, value_lib.Mirrored)
  
          run_options = config_pb2.RunOptions()
          run_options.experimental.collective_graph_key = 7
          if num_gpus > 1:
              result = sess.run(
                  [ops.convert_to_tensor(v) for v in result.values],
                  options=run_options,
              )[0]
          else:
              result = sess.run(
                  ops.convert_to_tensor(result), options=run_options
              )
  
          # Reduce the same indexed slices on CPU locally as our expected results.
          devices_cpu = [(worker_device or "") + "/device:CPU:0"] * (
              max(num_gpus, 1) * num_workers
          )
          per_replica_on_cpu = self._get_indexed_slices(
              devices_cpu, 0, variable_length, as_per_replica=False
          )
          expected_result = (
              cross_device_utils.aggregate_tensors_or_indexed_slices(
                  per_replica_on_cpu
              )
          )
          expected_result = sess.run(
              ops.convert_to_tensor(expected_result)
          )
  
          self.assertAllEqual(expected_result, result)
  
imports: import itertools
  import os
  import threading
  import time
  from absl.testing import parameterized
  import numpy
  from tensorflow.core.protobuf import config_pb2
  from tensorflow.python.distribute import cluster_resolver
  from tensorflow.python.distribute import collective_all_reduce_strategy
  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 distribute_utils
  from tensorflow.python.distribute import multi_worker_test_base
  from tensorflow.python.distribute import multi_worker_util
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import strategy_combinations
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.eager import test
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import indexed_slices
  from tensorflow.python.framework import kernels
  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 math_ops
  from tensorflow.python.ops import variables
label_desc: Use image node_id 7 for calling the CollectiveAllReduceTest obj's underlying member method code with example usage: obj._test_reduce_indexed_slices(task_type, task_id, num_gpus, communication, batch_reduce, variable_length, local_mode) without return types