diff --git a/.python-version b/.python-version index 24ee5b1..6324d40 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.13 +3.14 diff --git a/eris/__main__.py b/eris/__main__.py index 305d9dc..7c01a21 100755 --- a/eris/__main__.py +++ b/eris/__main__.py @@ -1266,6 +1266,9 @@ def inotify_watches_exceeded(): def entry_point(): + if "--worker" in sys.argv: + worker.main() + return (root_path, worker_count, editor_command, theme, compression, use_window) = check_arguments() manage_cache(root_path) diff --git a/eris/worker.py b/eris/worker.py index ef67296..37db9bf 100755 --- a/eris/worker.py +++ b/eris/worker.py @@ -1,6 +1,8 @@ import asyncio -import concurrent.futures +import contextlib +import os +import signal import sys import fill3 @@ -9,21 +11,10 @@ import eris.tools as tools import eris.paged_list -def _worker_run(path, tool_name, compression): - import os - import sys - with open(os.devnull, "w") as null: - old_out, old_err = sys.stdout, sys.stderr - sys.stdout = sys.stderr = null - try: - tool = getattr(tools, tool_name) - result = tools.Result(path, tool) - result.compression = compression - status, text = tools.run_tool_no_error(path, tool) - result.result = make_result_widget(text, result, compression) - return status.value - finally: - sys.stdout, sys.stderr = old_out, old_err +def worker_command(): + if getattr(sys, "frozen", False): + return [sys.executable, "--worker"] + return [sys.executable, os.path.abspath(sys.argv[0]), "--worker"] class Worker: @@ -34,15 +25,34 @@ class Worker: def __init__(self, compression): self.compression = compression self.result = None - self.executor = concurrent.futures.ProcessPoolExecutor(max_workers=1) + self.process = None + self.child_pgid = None + + async def create_process(self, command=None): + create = asyncio.create_subprocess_exec( + *(command or worker_command()), stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + preexec_fn=os.setsid) + self.process = await create + pid_line = await self.process.stdout.readline() + self.child_pgid = int(pid_line.strip()) + if "SNAP" not in os.environ: # Can't do setpriority inside snaps. + os.setpriority(os.PRIO_PGRP, self.child_pgid, 19) + self.process.stdin.write(f"{self.compression}\n".encode("utf-8")) async def run_tool(self, path, tool): - loop = asyncio.get_running_loop() - status_val = await loop.run_in_executor( - self.executor, _worker_run, path, tool.__qualname__, self.compression) - return tools.Status(status_val) + while True: + self.process.stdin.write(f"{tool.__qualname__}\n{path}\n" + .encode("utf-8", "surrogatepass")) + data = await self.process.stdout.readline() + if data == b"": + await self.create_process() + else: + break + return tools.Status(int(data)) async def job_runner(self, screen, summary, log, jobs_added_event): + await self.create_process() while True: await jobs_added_event.wait() while True: @@ -67,9 +77,9 @@ class Worker: jobs_added_event.clear() def kill(self): - for proc in self.executor._processes.values(): - proc.kill() - self.executor.shutdown(wait=False, cancel_futures=True) + if self.child_pgid is not None: + with contextlib.suppress(ProcessLookupError): + os.killpg(self.child_pgid, signal.SIGKILL) def make_result_widget(text, result, compression): @@ -80,3 +90,23 @@ def make_result_widget(text, result, compression): appearance, result.get_pages_dir(), page_size, cache_size=2, exist_ok=True, open_func=tools.compression_open_func(compression)) return fill3.Fixed(appearance) + + +def main(): + print(os.getpgid(os.getpid()), flush=True) + compression = input() + try: + while True: + tool_name, path = input(), input() + tool = getattr(tools, tool_name) + result = tools.Result(path, tool) + result.compression = compression + status, text = tools.run_tool_no_error(path, tool) + result.result = make_result_widget(text, result, compression) + print(status.value, flush=True) + except Exception: + tools.log_error() + + +if __name__ == "__main__": + main() diff --git a/tests/worker_test.py b/tests/worker_test.py index c9cad5d..451cdf3 100755 --- a/tests/worker_test.py +++ b/tests/worker_test.py @@ -2,6 +2,7 @@ import asyncio import os import shutil +import sys import tempfile import unittest @@ -28,6 +29,8 @@ class WorkerTestCase(unittest.TestCase): loop = asyncio.new_event_loop() compression = "none" self.worker_ = worker.Worker(compression) + command = [sys.executable, "-m", "eris.worker", "--worker"] + loop.run_until_complete(self.worker_.create_process(command=command)) future = self.worker_.run_tool("foo", tools.metadata) status = loop.run_until_complete(future) self.assertEqual(status, tools.Status.ok)