class_docstr: Manages running threads and processes of a Launchpad Program.
class_docstr_tok: ["Manages","running","threads","and","processes","of","a","Launchpad","Program","."]
docstr: Initializes a WorkerManager.
  
  Args:
    termination_notice_secs: 1) when >0, it's the countdown before a SIGKILL
      is issued upon a user-requested stop (relies on handle_user_stop=True).
      2) when =0, SIGKILL happens immediately upon user-requested stop.
    kill_all_upon_sigint: When True, set the stop event and kill all worker
      subprocesses, as well as the main process upon SIGINT. This allows the
      user to stop the program with Ctrl+C. It should be set to True for
      local_mp and local_mt.
    kill_workers_upon_sigint: When True, similar to kill_all_upon_sigint,
      except that it doesn't kill the main process. This is needed in test_mp
      so as not to fail the test upon cleanup.
    handle_sigterm: When this is True, kill all workers upon SIGTERM, by 1)
      forwarding SIGTERM to process workers 2) setting stop event for thread
      workers. Set this to True in process_entry.py so that the stop event
      will be triggered in the subprocesses via SIGTERM.
    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. It should be True for process_entry.py.
    process_tree_depth: the depth of Launchpad subprocesses in the process
      tree. For example, when the process is managed by GNOME, this value
      should be 2, so that in a tree of gnome-terminal -> bash -> interpreter
docstr_tok: ["Initializes","a","WorkerManager",".","Args",":","termination_notice_secs",":","1",")","when",">","0",",","it","'s","the","countdown","before","a","SIGKILL","is","issued","upon","a","user-requested","stop","(","relies","on","handle_user_stop=True",")",".","2",")","when","=0",",","SIGKILL","happens","immediately","upon","user-requested","stop",".","kill_all_upon_sigint",":","When","True",",","set","the","stop","event","and","kill","all","worker","subprocesses",",","as","well","as","the","main","process","upon","SIGINT",".","This","allows","the","user","to","stop","the","program","with","Ctrl+C",".","It","should","be","set","to","True","for","local_mp","and","local_mt",".","kill_workers_upon_sigint",":","When","True",",","similar","to","kill_all_upon_sigint",",","except","that","it","does","n't","kill","the","main","process",".","This","is","needed","in","test_mp","so","as","not","to","fail","the","test","upon","cleanup",".","handle_sigterm",":","When","this","is","True",",","kill","all","workers","upon","SIGTERM",",","by","1",")","forwarding","SIGTERM","to","process","workers","2",")","setting","stop","event","for","thread","workers",".","Set","this","to","True","in","process_entry.py","so","that","the","stop","event","will","be","triggered","in","the","subprocesses","via","SIGTERM",".","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",".","It","should","be","True","for","process_entry.py",".","process_tree_depth",":","the","depth","of","Launchpad","subprocesses","in","the","process","tree",".","For","example",",","when","the","process","is","managed","by","GNOME",",","this","value","should","be","2",",","so","that","in","a","tree","of","gnome-terminal","-",">","bash","-",">","interpreter"]
code: def __init__(
      self,
      termination_notice_secs: Optional[int] = None,
      kill_all_upon_sigint: bool = False,
      kill_workers_upon_sigint: bool = False,
      handle_sigterm: bool = False,
      register_in_thread: bool = False,
      process_tree_depth: int = 0,
  ):
      """Initializes a WorkerManager.
  
      Args:
        termination_notice_secs: 1) when >0, it's the countdown before a SIGKILL
          is issued upon a user-requested stop (relies on handle_user_stop=True).
          2) when =0, SIGKILL happens immediately upon user-requested stop.
        kill_all_upon_sigint: When True, set the stop event and kill all worker
          subprocesses, as well as the main process upon SIGINT. This allows the
          user to stop the program with Ctrl+C. It should be set to True for
          local_mp and local_mt.
        kill_workers_upon_sigint: When True, similar to kill_all_upon_sigint,
          except that it doesn't kill the main process. This is needed in test_mp
          so as not to fail the test upon cleanup.
        handle_sigterm: When this is True, kill all workers upon SIGTERM, by 1)
          forwarding SIGTERM to process workers 2) setting stop event for thread
          workers. Set this to True in process_entry.py so that the stop event
          will be triggered in the subprocesses via SIGTERM.
        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. It should be True for process_entry.py.
        process_tree_depth: the depth of Launchpad subprocesses in the process
          tree. For example, when the process is managed by GNOME, this value
          should be 2, so that in a tree of gnome-terminal -> bash -> interpreter
      """
      if termination_notice_secs is None:
          termination_notice_secs = (
              lp_flags.LP_TERMINATION_NOTICE_SECS.value
          )
      if termination_notice_secs < 0:
          raise ValueError("termination_notice_secs must be >= 0.")
      self._termination_notice_secs = termination_notice_secs
      if kill_all_upon_sigint and kill_workers_upon_sigint:
          raise ValueError(
              "Only one of kill_all_upon_sigint and kill_workers_upon_sigint can "
              "be True."
          )
      if kill_all_upon_sigint:
          _register_signal_handler(
              signal.SIGINT,
              functools.partial(self._handle_sigint, True),
          )
      elif kill_workers_upon_sigint:
          _register_signal_handler(
              signal.SIGINT,
              functools.partial(self._handle_sigint, False),
          )
  
      if handle_sigterm:
          _register_signal_handler(
              signal.SIGTERM,
              self._set_stop_event_and_terminate_process_workers,
          )
      self._stop_event = threading.Event()
      self._thread_workers: MutableMapping[
          str, List[ThreadWorker]
      ] = collections.defaultdict(list)
      self._process_workers: MutableMapping[
          str, List[psutil.Process]
      ] = collections.defaultdict(list)
      self._mutex = threading.Lock()
      if register_in_thread:
          _WORKER_MANAGERS.manager = self
      self._process_tree_depth = process_tree_depth
  
imports: import collections
  from concurrent import futures
  import dataclasses
  import functools
  import signal
  import subprocess
  import threading
  import time
  from typing import Any, Callable, Iterable, List, Mapping, MutableMapping, Optional, Sequence, Tuple
  from absl import logging
  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(termination_notice_secs, kill_all_upon_sigint, kill_workers_upon_sigint, handle_sigterm, register_in_thread, process_tree_depth)