code: def _testReductionAndBroadcast(self, cross_device_ops, devices):
      if context.num_gpus() < sum(
          1 for d in devices if "GPU" in d.upper()
      ):
          self.skipTest("Not enough GPUs")
  
      with self.cached_session() as sess:
          values = [
              constant_op.constant(float(d))
              for d in range(len(devices))
          ]
          per_replica = _make_per_replica(values, devices)
          mean = (len(devices) - 1.0) / 2.0
  
          values_2 = [
              constant_op.constant(d + 1.0) for d in range(len(devices))
          ]
          per_replica_2 = _make_per_replica(values_2, devices)
          mean_2 = mean + 1.0
  
          destination_mirrored = _fake_mirrored(1.0, devices)
          destination_different = _fake_mirrored(
              1.0, device_util.resolve(_cpu_device)
          )
          destination_str = device_util.resolve(_cpu_device)
  
          all_destinations = [
              destination_mirrored,
              destination_different,
              destination_str,
          ]
  
          # test reduce()
          for destinations in all_destinations:
              self._assert_mirrored_equal(
                  cross_device_ops.reduce(
                      reduce_util.ReduceOp.MEAN,
                      per_replica,
                      destinations=destinations,
                  ),
                  _fake_mirrored(mean, destinations),
                  sess,
              )
              self._assert_mirrored_equal(
                  cross_device_ops.reduce(
                      reduce_util.ReduceOp.MEAN,
                      per_replica_2,
                      destinations=destinations,
                  ),
                  _fake_mirrored(mean_2, destinations),
                  sess,
              )
              self._assert_mirrored_equal(
                  cross_device_ops.reduce(
                      reduce_util.ReduceOp.SUM,
                      per_replica,
                      destinations=destinations,
                  ),
                  _fake_mirrored(mean * len(devices), destinations),
                  sess,
              )
              self._assert_mirrored_equal(
                  cross_device_ops.reduce(
                      reduce_util.ReduceOp.SUM,
                      per_replica_2,
                      destinations=destinations,
                  ),
                  _fake_mirrored(mean_2 * len(devices), destinations),
                  sess,
              )
  
          # test batch_reduce()
          for d1, d2 in itertools.product(
              all_destinations, all_destinations
          ):
              self._assert_mirrored_equal(
                  cross_device_ops.batch_reduce(
                      reduce_util.ReduceOp.MEAN,
                      [(per_replica, d1), (per_replica_2, d2)],
                  ),
                  [
                      _fake_mirrored(mean, d1),
                      _fake_mirrored(mean_2, d2),
                  ],
                  sess,
              )
              self._assert_mirrored_equal(
                  cross_device_ops.batch_reduce(
                      reduce_util.ReduceOp.SUM,
                      [(per_replica, d1), (per_replica_2, d2)],
                  ),
                  [
                      _fake_mirrored(mean * len(devices), d1),
                      _fake_mirrored(mean_2 * len(devices), d2),
                  ],
                  sess,
              )
  
          # test broadcast()
          for destinations in all_destinations:
              self._assert_mirrored_equal(
                  cross_device_ops.broadcast(
                      constant_op.constant(1.0), destinations
                  ),
                  _fake_mirrored(1.0, destinations),
                  sess,
              )
  
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 3 for calling the CrossDeviceOpsTestBase obj's underlying member method code with example usage: obj._testReductionAndBroadcast(cross_device_ops, devices) without return types