class_docstr: Manages running threads and processes of a Launchpad Program.
class_docstr_tok: ["Manages","running","threads","and","processes","of","a","Launchpad","Program","."]
docstr: Update self._thread_workers and return a tuple representing the change.
  
  This will update self._thread_workers so that it only contains active
  workers.
  
  Returns:
    A tuple. The first element of the tuple are futures for recently finished
    workers, and the second is a bool indicating if there are still active
    workers.
docstr_tok: ["Update","self._thread_workers","and","return","a","tuple","representing","the","change",".","This","will","update","self._thread_workers","so","that","it","only","contains","active","workers",".","Returns",":","A","tuple",".","The","first","element","of","the","tuple","are","futures","for","recently","finished","workers",",","and","the","second","is","a","bool","indicating","if","there","are","still","active","workers","."]
code: def _update_and_get_recently_finished(
      self,
  ) -> Tuple[List[futures.Future[Any]], bool]:
      """Update self._thread_workers and return a tuple representing the change.
  
      This will update self._thread_workers so that it only contains active
      workers.
  
      Returns:
        A tuple. The first element of the tuple are futures for recently finished
        workers, and the second is a bool indicating if there are still active
        workers.
      """
      recently_finished = []
      has_active_workers = False
      with self._mutex:
          active_threads = collections.defaultdict(list)
          for label in self._thread_workers:
              for worker in self._thread_workers[label]:
                  if worker.thread.is_alive():
                      active_threads[label].append(worker)
                      has_active_workers = True
                  else:
                      recently_finished.append(worker.future)
          self._thread_workers = active_threads
  
          active_processes = collections.defaultdict(list)
          for label, processes in self._process_workers.items():
              for process in processes:
                  if (
                      process.is_running()
                      and process.status() != psutil.STATUS_ZOMBIE
                  ):
                      has_active_workers = True
                      active_processes[label].append(process)
                  else:
                      future = futures.Future()
                      res = process.wait()
                      if res and not self._stop_event.is_set():
                          # Here we make sure stop_event hasn't been set yet, before we
                          # propagate the non-zero exit code. This is because when we
                          # forcefully terminate the program (e.g., due to `lp.stop()`),
                          # some subprocesses might have non-zero exit code.
                          future.set_exception(
                              RuntimeError(
                                  f'A "{label}" worker exited with code {res}.'
                              )
                          )
                      else:
                          future.set_result(None)
                      recently_finished.append(future)
          self._process_workers = active_processes
  
      return recently_finished, has_active_workers
  
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 8 for calling the WorkerManager obj's underlying member method code with example usage: obj._update_and_get_recently_finished() and returns: recently_finished,  has_active_workers