diff --git a/eris/__main__.py b/eris/__main__.py index 47ba88e..859c6a1 100755 --- a/eris/__main__.py +++ b/eris/__main__.py @@ -791,8 +791,7 @@ class Screen: self._log.log_message([in_green("Editing "), path_colored, in_green(f" at line {line_num}…")]) subprocess.Popen(f"{self.editor_command} +{line_num} {path}", shell=True, - stdout=subprocess.PIPE, stderr=subprocess.PIPE, - start_new_session=True) + stdout=subprocess.PIPE, stderr=subprocess.PIPE) def patch_file(self): result = self._summary.get_selection() if hasattr(result.tool, "patch_command") and result.tool.patch_command is not None: @@ -850,8 +849,7 @@ class Screen: path = self._summary.get_selection().path path_colored = lscolors.path_colored(path) self._log.log_message([in_green("Opening "), path_colored, in_green("…")]) - subprocess.Popen(["xdg-open", path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, - start_new_session=True) + subprocess.Popen(["xdg-open", path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) def save(self): worker.Worker.unsaved_jobs_total = 0 @@ -1079,10 +1077,6 @@ async def main(title, root_path, worker_count=None, editor_command=None, theme=N notifier.stop() if summary.is_loaded: screen.save() - if sys.stdout.isatty() and (os.getpgrp() == os.getpid() - or os.getpgrp() == os.getppid()): - sys.stdout.flush() - os.killpg(os.getpgrp(), signal.SIGKILL) @contextlib.contextmanager diff --git a/eris/worker.py b/eris/worker.py index 689ada1..3542cc5 100755 --- a/eris/worker.py +++ b/eris/worker.py @@ -1,7 +1,10 @@ import asyncio import concurrent.futures +import os +import signal import sys +import threading import fill3 @@ -17,12 +20,17 @@ class Worker: def __init__(self, compression): self.compression = compression self.result = None + self._thread_native_id = None + self._killed = False self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) def _run_job(self, path, tool): + self._thread_native_id = threading.current_thread().native_id result = tools.Result(path, tool) result.compression = self.compression status, text = tools.run_tool_no_error(path, tool) + if self._killed: + return status result.result = make_result_widget(text, result, self.compression) return status @@ -55,7 +63,36 @@ class Worker: jobs_added_event.clear() def kill(self): + self._killed = True self._executor.shutdown(wait=False, cancel_futures=True) + if self._thread_native_id is not None: + for child in self._thread_children(self._thread_native_id): + self._kill_descendants(child) + + @staticmethod + def _thread_children(tid): + try: + with open(f"/proc/{os.getpid()}/task/{tid}/children") as children_file: + return [int(child) for child in children_file.read().split()] + except OSError: + return [] + + @staticmethod + def _kill_descendants(pid): + descendants = [pid] + while descendants: + current = descendants.pop() + try: + task_dir = f"/proc/{current}/task" + for tid in os.listdir(task_dir): + with open(f"{task_dir}/{tid}/children") as children_file: + descendants.extend(int(child) for child in children_file.read().split()) + except OSError: + pass + try: + os.kill(current, signal.SIGKILL) + except ProcessLookupError: + pass def make_result_widget(text, result, compression): diff --git a/tests/worker_test.py b/tests/worker_test.py index c9cad5d..7347222 100755 --- a/tests/worker_test.py +++ b/tests/worker_test.py @@ -34,6 +34,47 @@ class WorkerTestCase(unittest.TestCase): result_path = os.path.join(tools.CACHE_PATH, "foo-metadata") self.assertTrue(os.path.exists(result_path)) + def test_kill_terminates_in_flight_subprocess(self): + import subprocess + import threading + import time + + loop = asyncio.new_event_loop() + self.worker_ = worker.Worker("none") + process_holder = {} + original_run_tool_no_error = tools.run_tool_no_error + + def slow_tool(path, tool): + process = subprocess.Popen(["sleep", "60"]) + process_holder["process"] = process + process.wait() + return tools.Status.ok, "" + + tools.run_tool_no_error = slow_tool + try: + results = [] + + def run_future(): + results.append(loop.run_until_complete(self.worker_.run_tool("foo", tools.metadata))) + + future_thread = threading.Thread(target=run_future) + future_thread.start() + deadline = time.time() + 5 + while "process" not in process_holder and time.time() < deadline: + time.sleep(0.01) + self.assertIn("process", process_holder) + start = time.time() + self.worker_.kill() + future_thread.join(timeout=5) + self.assertLess(time.time() - start, 5) + self.assertFalse(future_thread.is_alive()) + self.assertEqual(len(results), 1) + self.assertIsNotNone(process_holder["process"].poll()) + result_path = os.path.join(tools.CACHE_PATH, "foo-metadata") + self.assertFalse(os.path.exists(result_path)) + finally: + tools.run_tool_no_error = original_run_tool_no_error + if __name__ == "__main__": unittest.main()