docstr: Batch reduce the given `inputs` and verify the output matches `expect`.
  
  Args:
    inputs: a 2-level nested list of `Tensor` or `IndexedSlices`, where i-th
      value will be fed to i-th replica.
    expect: a list of `Tensor` or `IndexedSlices`. This should be the expected
      value for one replica.
    options: a `RunOpotions` instance.
docstr_tok: ["Batch","reduce","the","given","`","inputs","`","and","verify","the","output","matches","`","expect","`",".","Args",":","inputs",":","a","2-level","nested","list","of","`","Tensor","`","or","`","IndexedSlices","`",",","where","i-th","value","will","be","fed","to","i-th","replica",".","expect",":","a","list","of","`","Tensor","`","or","`","IndexedSlices","`",".","This","should","be","the","expected","value","for","one","replica",".","options",":","a","`","RunOpotions","`","instance","."]
code: def batch_reduce_and_verify(self, inputs, expect, options):
      """Batch reduce the given `inputs` and verify the output matches `expect`.
  
      Args:
        inputs: a 2-level nested list of `Tensor` or `IndexedSlices`, where i-th
          value will be fed to i-th replica.
        expect: a list of `Tensor` or `IndexedSlices`. This should be the expected
          value for one replica.
        options: a `RunOpotions` instance.
      """
  
      def replica_fn():
          CollectiveReplicaLauncher._prefer_unique_instance_key = (
              options.prefer_unique_instance_key
          )
          collective, devices, pid = self.make_collective(
              options.num_processes, options.gpus_per_process
          )
  
          def batch_reduce_fn():
              batch_size = len(inputs[0])
              value_dst_pairs = []
              for i in range(batch_size):
  
                  def value_fn(device_idx, idx=i):
                      return inputs[pid * len(devices) + device_idx][
                          idx
                      ]
  
                  per_replica_value = make_per_replica_value(
                      value_fn, devices
                  )
                  value_dst_pairs.append(
                      (per_replica_value, per_replica_value)
                  )
              reduced_values = collective.batch_reduce(
                  options.reduce_op,
                  value_dst_pairs,
                  options.communication_options,
              )
              if options.gpus_per_process > 1:
                  for v in reduced_values:
                      self.assertIsInstance(v, value_lib.Mirrored)
              reduced_values = [self.as_list(v) for v in reduced_values]
              for v in reduced_values:
                  self.assertAllEqual(devices, [t.device for t in v])
              return nest.map_structure(
                  ops.convert_to_tensor, reduced_values
              )
  
          per_replica_expect = nest.map_structure(
              lambda x: [ops.convert_to_tensor(x)] * len(devices),
              expect,
          )
  
          if "eager" in options.mode:
              got = batch_reduce_fn()
              self.assertAllClose(got, per_replica_expect)
  
          if "func_graph" in options.mode:
              got = def_function.function(batch_reduce_fn)()
              self.assertAllClose(got, per_replica_expect)
  
      get_global_mpr(options.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 5 for calling the CollectiveOpsTest obj's underlying member method code with example usage: obj.batch_reduce_and_verify(inputs, expect, options) and returns: nest, inputs