class_docstr: Tests distributed training.
class_docstr_tok: ["Tests","distributed","training","."]
docstr: Uses multiprocessing to simulate a distributed training environment.
docstr_tok: ["Uses","multiprocessing","to","simulate","a","distributed","training","environment","."]
code: def test_distributed_training(
      self,
      num_workers,
      num_ps,
      placement_strategy,
      estimator="estimator",
  ):
      """Uses multiprocessing to simulate a distributed training environment."""
  
      # Inspired by `tf.test.create_local_cluster`.
      worker_ports = [_pick_unused_port() for _ in range(num_workers)]
      ps_ports = [_pick_unused_port() for _ in range(num_ps)]
      ws_targets = ["localhost:%s" % port for port in worker_ports]
      ps_targets = ["localhost:%s" % port for port in ps_ports]
  
      # For details see:
      # https://www.tensorflow.org/api_docs/python/tf/estimator/train_and_evaluate
      tf_config = {
          "cluster": {
              # The chief is always worker 0.
              "chief": [ws_targets[0]],
          },
          "task": {"type": "chief", "index": 0},
      }
  
      # The chief is already worker 0.
      if len(ws_targets) > 1:
          tf_config["cluster"]["worker"] = ws_targets[1:]
      if ps_targets:
          tf_config["cluster"]["ps"] = ps_targets
  
      worker_processes = []
      ps_processes = []
      evaluator_processes = []
  
      model_dir = self.test_subdirectory
  
      # Chief
      worker_processes.append(
          _create_task_process(
              "chief",
              0,
              estimator,
              placement_strategy,
              tf_config,
              model_dir,
          )
      )
      # Workers
      for i in range(len(ws_targets[1:])):
          worker_processes.append(
              _create_task_process(
                  "worker",
                  i,
                  estimator,
                  placement_strategy,
                  tf_config,
                  model_dir,
              )
          )
      # Parameter Servers (PS)
      for i in range(len(ps_targets)):
          ps_processes.append(
              _create_task_process(
                  "ps",
                  i,
                  estimator,
                  placement_strategy,
                  tf_config,
                  model_dir,
              )
          )
      # Evaluator
      evaluator_processes.append(
          _create_task_process(
              "evaluator",
              0,
              estimator,
              placement_strategy,
              tf_config,
              model_dir,
          )
      )
  
      # Run processes.
      try:
          # NOTE: Parameter servers do not shut down on their own.
          self._wait_for_processes(
              worker_processes + evaluator_processes,
              kill_processes=ps_processes,
              timeout_secs=500,
          )
      finally:
          for process in (
              worker_processes + ps_processes + evaluator_processes
          ):
              try:
                  process.popen.kill()
              except OSError:
                  pass  # It's OK (and expected) if the process already exited.
              process.stderr.close()
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  import copy
  import itertools
  import json
  import os
  import shutil
  import socket
  import subprocess
  import sys
  import time
  from absl import flags
  from absl import logging
  from absl.testing import parameterized
  from adanet import tf_compat
  from adanet.core.timer import _CountDownTimer
  import tensorflow.compat.v2
label_desc: Use image node_id 3 for calling the EstimatorDistributedTrainingTest obj's underlying member method code with example usage: obj.test_distributed_training(num_workers, num_ps, placement_strategy, estimator) without return types