class_docstr: Tests tf.data service reading from local or non-TPU workers.
  
  When `target_workers` is "AUTO", tf.data service avoids cross-TPU reads, to
  avoid RPCS and data serialization / deserialization, thus improving resource
  utilization of TPU hosts.
class_docstr_tok: ["Tests","tf.data","service","reading","from","local","or","non-TPU","workers",".","When","`","target_workers","`","is","``","AUTO","''",",","tf.data","service","avoids","cross-TPU","reads",",","to","avoid","RPCS","and","data","serialization","\/","deserialization",",","thus","improving","resource","utilization","of","TPU","hosts","."]
code: def testAddRemoteWorkersMidJob(
      self, num_local_workers, num_remote_workers
  ):
      cluster = multi_process_cluster.MultiProcessCluster(
          num_local_workers=num_local_workers,
          num_remote_workers=num_remote_workers,
          worker_tags=[_COLOCATED_WORKER_TAG],
      )
  
      # num_elements needs to be bigger than (100 + <cpu core count>), the extra
      # 100 is just a bit of margin. The CPU core count is involved as
      # elements are prefetched, one element per CPU core.
      num_elements = 200 + multiprocessing.cpu_count()
      dataset = self.make_distributed_range_dataset(
          num_elements, cluster
      )
      get_next = self.getNext(dataset)
      results = [self.evaluate(get_next()) for _ in range(100)]
  
      # Will only read from the two non-TPU workers.
      cluster.start_remote_worker(worker_tags=None)
      cluster.start_remote_worker(worker_tags=[_COLOCATED_WORKER_TAG])
      cluster.start_remote_worker(worker_tags=None)
      cluster.start_remote_worker(worker_tags=[_COLOCATED_WORKER_TAG])
      expect_num_workers_to_read = num_local_workers + 2
  
      # Wait for the new worker to register with the dispatcher.
      while cluster._dispatcher._num_workers() < (
          num_local_workers + num_remote_workers + 4
      ):
          time.sleep(10 / 1000)  # 10ms
  
      results += self.getIteratorOutput(get_next)
      self.assertCountEqual(
          results,
          expect_num_workers_to_read * list(range(num_elements)),
      )
  
imports: import time
  import multiprocessing
  from absl.testing import parameterized
  from tensorflow.python.data.experimental.kernel_tests.service import multi_process_cluster
  from tensorflow.python.data.experimental.kernel_tests.service import test_base
  from tensorflow.python.data.experimental.ops import data_service_ops
  from tensorflow.python.data.kernel_tests import test_base
  from tensorflow.python.framework import combinations
label_desc: Use image node_id 7 for calling the WorkerTagsTest obj's underlying member method code with example usage: obj.testAddRemoteWorkersMidJob(num_local_workers, num_remote_workers) without return types