class_docstr: Manages running threads and processes of a Launchpad Program.
class_docstr_tok: ["Manages","running","threads","and","processes","of","a","Launchpad","Program","."]
docstr: Handles SIGINT.
  
  This does the following:
  
  1. Set the stop event. Nodes can listen to the stop event and perform
     cleanup actions.
  2. Wait for termination_notice_secs (specified from   __init__()`), since
     the workers might need some time for cleanup.
  3. SIGKILL the remaining workers.
  
  if kill_main_process=True, also kill the main process. This should be set
  True for local_mp, but not for test_mp as it will fail the test case.
  
  Args:
    kill_main_process: whether or not to kill the main process after killing
      all the subprocesses.
docstr_tok: ["Handles","SIGINT",".","This","does","the","following",":","1",".","Set","the","stop","event",".","Nodes","can","listen","to","the","stop","event","and","perform","cleanup","actions",".","2",".","Wait","for","termination_notice_secs","(","specified","from","__init__","(",")","`",")",",","since","the","workers","might","need","some","time","for","cleanup",".","3",".","SIGKILL","the","remaining","workers",".","if","kill_main_process=True",",","also","kill","the","main","process",".","This","should","be","set","True","for","local_mp",",","but","not","for","test_mp","as","it","will","fail","the","test","case",".","Args",":","kill_main_process",":","whether","or","not","to","kill","the","main","process","after","killing","all","the","subprocesses","."]
code: def _handle_sigint(self, kill_main_process: bool):
      """Handles SIGINT.
  
      This does the following:
  
      1. Set the stop event. Nodes can listen to the stop event and perform
         cleanup actions.
      2. Wait for termination_notice_secs (specified from   __init__()`), since
         the workers might need some time for cleanup.
      3. SIGKILL the remaining workers.
  
      if kill_main_process=True, also kill the main process. This should be set
      True for local_mp, but not for test_mp as it will fail the test case.
  
      Args:
        kill_main_process: whether or not to kill the main process after killing
          all the subprocesses.
      """
      print(
          termcolor.colored(
              "User-requested termination. Asking workers to stop.",
              "blue",
          )
      )
      # Notify all the thread workers.
      self._stop_event.set()
  
      # Notify all the process workers.
      for _, processes in self._process_workers.items():
          _send_signal_to_processes_with_depth(
              processes, signal.SIGTERM, self._process_tree_depth
          )
  
      def _force_stop(kill_main_process=kill_main_process):
          # Since we are forcefully stopping the system, we send signals to all
          # levels of the process trees. This makes sure to kill
          # tmux/gnome-terminal/etc, processes that create the Launchpad
          # subprocesses.
          for _, processes in self._process_workers.items():
              for process in processes:
                  for child in process.children(recursive=True):
                      child.send_signal(signal.SIGKILL)
                  process.send_signal(signal.SIGKILL)
          if kill_main_process:
              signal.raise_signal(signal.SIGKILL)
  
      if self._termination_notice_secs > 0:
          print(
              termcolor.colored(
                  "Press CTRL+C to terminate immediately.", "blue"
              )
          )
  
          _register_signal_handler(signal.SIGINT, _force_stop)
          pending_secs = self._termination_notice_secs
          while self._has_active_workers() and pending_secs:
              print(
                  termcolor.colored(
                      f"Waiting for workers to stop for {pending_secs}s.",
                      "blue",
                  ),
                  end="\r",
              )
              time.sleep(1)
              pending_secs -= 1
      if self._has_active_workers():
          print(termcolor.colored("\nKilling entire runtime.", "blue"))
          _force_stop()
  
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 12 for calling the WorkerManager obj's underlying member method code with example usage: obj._handle_sigint(kill_main_process) without return types