airflow/utils/dag_processing.py
def end(self):
        """
        Kill all child processes on exit since we don't want to leave
        them as orphaned.
        """
        pids_to_kill = self.get_all_pids()
        if len(pids_to_kill) > 0:
            # First try SIGTERM
            this_process = psutil.Process(os.getpid())
            # Only check child processes to ensure that we don't have a case
            # where we kill the wrong process because a child process died
            # but the PID got reused.
            child_processes = [x for x in this_process.children(recursive=True)
                               if x.is_running() and x.pid in pids_to_kill]
            for child in child_processes:
                self.log.info("Terminating child PID: %s", child.pid)
                child.terminate()
            # TODO: Remove magic number
            timeout = 5
            self.log.info("Waiting up to %s seconds for processes to exit...", timeout)
            try:
                psutil.wait_procs(
                    child_processes, timeout=timeout,
                    callback=lambda x: self.log.info('Terminated PID %s', x.pid))
            except psutil.TimeoutExpired:
                self.log.debug("Ran out of time while waiting for processes to exit")

            # Then SIGKILL
            child_processes = [x for x in this_process.children(recursive=True)
                               if x.is_running() and x.pid in pids_to_kill]
            if len(child_processes) > 0:
                self.log.info("SIGKILL processes that did not terminate gracefully")
                for child in child_processes:
                    self.log.info("Killing child PID: %s", child.pid)
                    child.kill()
                    child.wait()
def end(self):
        pids_to_kill = self.get_all_pids()
        if len(pids_to_kill) > 0:
            # First try SIGTERM
            this_process = psutil.Process(os.getpid())
            # Only check child processes to ensure that we don't have a case
            # where we kill the wrong process because a child process died
            # but the PID got reused.
            child_processes = [x for x in this_process.children(recursive=True)
                               if x.is_running() and x.pid in pids_to_kill]
            for child in child_processes:
                self.log.info("Terminating child PID: %s", child.pid)
                child.terminate()
            # TODO: Remove magic number
            timeout = 5
            self.log.info("Waiting up to %s seconds for processes to exit...", timeout)
            try:
                psutil.wait_procs(
                    child_processes, timeout=timeout,
                    callback=lambda x: self.log.info('Terminated PID %s', x.pid))
            except psutil.TimeoutExpired:
                self.log.debug("Ran out of time while waiting for processes to exit")

            # Then SIGKILL
            child_processes = [x for x in this_process.children(recursive=True)
                               if x.is_running() and x.pid in pids_to_kill]
            if len(child_processes) > 0:
                self.log.info("SIGKILL processes that did not terminate gracefully")
                for child in child_processes:
                    self.log.info("Killing child PID: %s", child.pid)
                    child.kill()
                    child.wait()
def end ( self ) : pids_to_kill = self . get_all_pids ( ) if len ( pids_to_kill ) > 0 : # First try SIGTERM this_process = psutil . Process ( os . getpid ( ) ) # Only check child processes to ensure that we don't have a case # where we kill the wrong process because a child process died # but the PID got reused. child_processes = [ x for x in this_process . children ( recursive = True ) if x . is_running ( ) and x . pid in pids_to_kill ] for child in child_processes : self . log . info ( "Terminating child PID: %s" , child . pid ) child . terminate ( ) # TODO: Remove magic number timeout = 5 self . log . info ( "Waiting up to %s seconds for processes to exit..." , timeout ) try : psutil . wait_procs ( child_processes , timeout = timeout , callback = lambda x : self . log . info ( 'Terminated PID %s' , x . pid ) ) except psutil . TimeoutExpired : self . log . debug ( "Ran out of time while waiting for processes to exit" ) # Then SIGKILL child_processes = [ x for x in this_process . children ( recursive = True ) if x . is_running ( ) and x . pid in pids_to_kill ] if len ( child_processes ) > 0 : self . log . info ( "SIGKILL processes that did not terminate gracefully" ) for child in child_processes : self . log . info ( "Killing child PID: %s" , child . pid ) child . kill ( ) child . wait ( )
Kill all child processes on exit since we don't want to leave
        them as orphaned.
Kill all child processes on exit since we don t want to leave them as orphaned .