parameters: (self)
docstring: Observer thread for ProcPool used for stopping and joining processes.
docstring_summary: Observer thread for ProcPool used for stopping and joining processes.
docstring_tokens: Observer thread for ProcPool used for stopping and joining processes .
function: def _observer_thread(self):
          """Observer thread for ProcPool used for stopping and joining processes.
          """
          exit_gently = True
          try:
              ps = {p.sentinel: p for p in self._processes}
              listen_for = list(ps.keys()) + [self._interruption_pipe]
              # Once one process exits stop the whole group (gracefully if possible)
              while True:
                  sentinels = multiprocessing.connection.wait(listen_for)
                  proc_sentinels = [s for s in sentinels if s != self._interruption_pipe]
                  if self._interruption_pipe in sentinels:
                      break
                  if any(ps[sentinel].exitcode is not None for sentinel in proc_sentinels):
                      exit_gently = False
                      break
          except:
              exit_gently = False
              raise
          finally:
              self._result_queue.close()  # main thread may be blocked on reading the queue
              if exit_gently:
                  # try to close task queues and notify waiting processes, so that they can
                  # cleanup and exit. Unfortunately if all workers exited abruptly when waiting,
                  # an attempt to notify workers with multiprocessing.Condition might lead to deadlock on
                  # underlying semaphore. For this reason it is done only if none of the workers reported
                  # to have exited.
                  for queue in self._task_queues:
                      queue.close()
                  for proc in self._processes:
                      proc.join(1)
              for proc in self._processes:
                  if proc.exitcode is None:
                      proc.terminate()
                      proc.join()
function_tokens: def _observer_thread ( self ) : exit_gently = True try : ps = { p . sentinel : p for p in self . _processes } listen_for = list ( ps . keys ( ) ) + [ self . _interruption_pipe ] # Once one process exits stop the whole group (gracefully if possible) while True : sentinels = multiprocessing . connection . wait ( listen_for ) proc_sentinels = [ s for s in sentinels if s != self . _interruption_pipe ] if self . _interruption_pipe in sentinels : break if any ( ps [ sentinel ] . exitcode is not None for sentinel in proc_sentinels ) : exit_gently = False break except : exit_gently = False raise finally : self . _result_queue . close ( ) # main thread may be blocked on reading the queue if exit_gently : # try to close task queues and notify waiting processes, so that they can # cleanup and exit. Unfortunately if all workers exited abruptly when waiting, # an attempt to notify workers with multiprocessing.Condition might lead to deadlock on # underlying semaphore. For this reason it is done only if none of the workers reported # to have exited. for queue in self . _task_queues : queue . close ( ) for proc in self . _processes : proc . join ( 1 ) for proc in self . _processes : if proc . exitcode is None : proc . terminate ( ) proc . join ( )