class_docstr: Encapsulates running threads and processes of a Launchpad Program.
class_docstr_tok: ["Encapsulates","running","threads","and","processes","of","a","Launchpad","Program","."]
docstr: Initializes a WorkerManager.
  
  Args:
    kill_main_thread: When True (the default), it will kill the current
      process (program launcher or the node) after killing all th
      subprocesses, to guarantee complete cleanup of the program. Setting it
      to False disables this behavior, which can be useful in test_mp, where
      killing the main process causes a test to fail.
    register_in_thread: Make the worker manager accessible through
      `get_worker_manager()` in the current thread (needed by `stop_event()`
      for example). It should be False if we don't need to access
      `get_worker_manager()` , e.g. at the launcher thread of local_mt and
      local_mp.
docstr_tok: ["Initializes","a","WorkerManager",".","Args",":","kill_main_thread",":","When","True","(","the","default",")",",","it","will","kill","the","current","process","(","program","launcher","or","the","node",")","after","killing","all","th","subprocesses",",","to","guarantee","complete","cleanup","of","the","program",".","Setting","it","to","False","disables","this","behavior",",","which","can","be","useful","in","test_mp",",","where","killing","the","main","process","causes","a","test","to","fail",".","register_in_thread",":","Make","the","worker","manager","accessible","through","`","get_worker_manager","(",")","`","in","the","current","thread","(","needed","by","`","stop_event","(",")","`","for","example",")",".","It","should","be","False","if","we","do","n't","need","to","access","`","get_worker_manager","(",")","`",",","e.g",".","at","the","launcher","thread","of","local_mt","and","local_mp","."]
code: def __init__(
      self,
      kill_main_thread=True,
      register_in_thread=False,
  ):
      """Initializes a WorkerManager.
  
      Args:
        kill_main_thread: When True (the default), it will kill the current
          process (program launcher or the node) after killing all th
          subprocesses, to guarantee complete cleanup of the program. Setting it
          to False disables this behavior, which can be useful in test_mp, where
          killing the main process causes a test to fail.
        register_in_thread: Make the worker manager accessible through
          `get_worker_manager()` in the current thread (needed by `stop_event()`
          for example). It should be False if we don't need to access
          `get_worker_manager()` , e.g. at the launcher thread of local_mt and
          local_mp.
      """
      self._mutex = threading.Lock()
      self._termination_notice_secs = -1
      handle_user_stop = True
      global _HAS_MAIN_MANAGER
      if not _HAS_MAIN_MANAGER:
          # This logic resolves the potential conflict between two WorkerManagers
          # in the same process. In particular, only the first one will execute the
          # "countdown-and-sigkill" logic upon Ctrl+C.
          self._termination_notice_secs = (
              FLAGS.lp_termination_notice_secs
          )
          _HAS_MAIN_MANAGER = True
      self._active_workers = collections.defaultdict(list)
      self._workers_count = collections.defaultdict(lambda: 0)
      self._first_failure = None
      self._stop_counter = 0
      self._kill_main_thread = kill_main_thread
      self._stop_event = threading.Event()
      self._main_thread = threading.current_thread().ident
      _register_signal_handler(signal.SIGTERM, self._sigterm)
      _register_signal_handler(signal.SIGQUIT, self._sigquit)
      if handle_user_stop:
          _register_signal_handler(signal.SIGINT, self._stop_by_user)
      if register_in_thread:
          _WORKER_MANAGERS.manager = self
  
imports: import atexit
  import collections
  from concurrent import futures
  import os
  import signal
  import subprocess
  import sys
  import threading
  import time
  import traceback
  from typing import Optional, Sequence, Text
  from absl import flags
  from absl import logging
  from absl.testing import absltest
  from launchpad import flags
  import psutil
  import termcolor
label_desc: Use image node_id 1 to create a new WorkerManager object  with example: obj = WorkerManager(kill_main_thread, register_in_thread)