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):
        """
        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()
Kill all child processes on exit since we don't want to leave
        them as orphaned.