class_docstr: Tests distributed training.
class_docstr_tok: ["Tests","distributed","training","."]
docstr: Waits until all `wait_processes` finish, then kills `kill_processes`.
  
  Fails an assert if a process in `wait_processes` finishes unsuccessfully.
  The processes in `kill_processes` are assumed to never finish so they are
  killed.
  
  Args:
    wait_processes: A list of _ProcessInfo tuples. This function will wait for
      each to finish.
    kill_processes: A list of _ProcessInfo tuples. Each will be killed once
      every process in `wait_processes` is finished.
    timeout_secs: Seconds to wait before timing out and terminating processes.
  
  Returns:
    A list of strings, each which is a string of the stderr of a wait process.
  
  Raises:
    Exception: When waiting for tasks to finish times out.
docstr_tok: ["Waits","until","all","`","wait_processes","`","finish",",","then","kills","`","kill_processes","`",".","Fails","an","assert","if","a","process","in","`","wait_processes","`","finishes","unsuccessfully",".","The","processes","in","`","kill_processes","`","are","assumed","to","never","finish","so","they","are","killed",".","Args",":","wait_processes",":","A","list","of","_ProcessInfo","tuples",".","This","function","will","wait","for","each","to","finish",".","kill_processes",":","A","list","of","_ProcessInfo","tuples",".","Each","will","be","killed","once","every","process","in","`","wait_processes","`","is","finished",".","timeout_secs",":","Seconds","to","wait","before","timing","out","and","terminating","processes",".","Returns",":","A","list","of","strings",",","each","which","is","a","string","of","the","stderr","of","a","wait","process",".","Raises",":","Exception",":","When","waiting","for","tasks","to","finish","times","out","."]
code: def _wait_for_processes(
      self, wait_processes, kill_processes, timeout_secs
  ):
      """Waits until all `wait_processes` finish, then kills `kill_processes`.
  
      Fails an assert if a process in `wait_processes` finishes unsuccessfully.
      The processes in `kill_processes` are assumed to never finish so they are
      killed.
  
      Args:
        wait_processes: A list of _ProcessInfo tuples. This function will wait for
          each to finish.
        kill_processes: A list of _ProcessInfo tuples. Each will be killed once
          every process in `wait_processes` is finished.
        timeout_secs: Seconds to wait before timing out and terminating processes.
  
      Returns:
        A list of strings, each which is a string of the stderr of a wait process.
  
      Raises:
        Exception: When waiting for tasks to finish times out.
      """
  
      timer = _CountDownTimer(timeout_secs)
      finished_wait_processes = set()
      poll_count = {
          wait_process: 0.0 for wait_process in wait_processes
      }
  
      while len(finished_wait_processes) < len(wait_processes):
          if timer.secs_remaining() == 0:
              logging.error(
                  "Timed out! Outputting logs of unfinished processes:"
              )
              for i, wait_process in enumerate(wait_processes):
                  if i in finished_wait_processes:
                      continue
                  log_all(wait_process, "incompleted")
              raise Exception(
                  "Timed out waiting for tasks to complete."
              )
          for i, wait_process in enumerate(wait_processes):
              if i in finished_wait_processes:
                  continue
              ret_code = wait_process.popen.poll()
              if ret_code is None:
                  poll_count[wait_process] += 0.25
                  if (
                      (poll_count[wait_process] / 10.0)
                      - int(poll_count[wait_process] / 10.0)
                  ) == 0:
                      logging.info(
                          "%d secs has elapsed for %s",
                          poll_count[wait_process],
                          wait_process.name,
                      )
                  continue
              logging.info("%s finished", wait_process.name)
              log_all(wait_process, "completed")
              self.assertEqual(0, ret_code)
              finished_wait_processes.add(i)
          for kill_process in kill_processes:
              ret_code = kill_process.popen.poll()
              # Kill processes should not end until we kill them.
              # If it returns early, note the return code.
              if ret_code is not None:
                  logging.error(
                      "kill process %s ended with ret_code %d",
                      kill_process.name,
                      ret_code,
                  )
                  log_all(
                      kill_process,
                      "ended with code {}".format(ret_code),
                  )
                  self.assertIsNone(ret_code)
          # Delay between polling loops.
          time.sleep(0.25)
      logging.info("All wait processes finished")
      for i, kill_process in enumerate(kill_processes):
          # Kill each kill process.
          kill_process.popen.kill()
          kill_process.popen.wait()
          log_all(kill_process, "killed")
  
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 2 for calling the EstimatorDistributedTrainingTest obj's underlying member method code with example usage: obj._wait_for_processes(wait_processes, kill_processes, timeout_secs) without return types