replace eris-worker subprocess with ProcessPoolExecutor; guard sweep generators against empty _entries

- Rewrite worker.py: each Worker uses a ProcessPoolExecutor(max_workers=1),
  tools run via loop.run_in_executor() in a forked OS subprocess
- Remove create_process(), main(), stdin/stdout IPC, os.killpg
- Remove eris-worker entry point from [project.scripts] in pyproject.toml
- Update tests/worker_test.py for new Worker API
- Guard _sweep_up/_sweep_down against IndexError when _entries is empty;
  keep cursor_position() try/except for stale y after row deletion

eris/__main__.py

@@ -370,12 +370,16 @@ class Summary:
         log.log_message(f"Finished sync with filesystem. {round(duration, 2)} secs")
 
     def _sweep_up(self, x, y):
+        if not self._entries:
+            return
         yield from reversed(self._entries[y][:x])
         while True:
             y = (y - 1) % len(self._entries)
             yield from reversed(self._entries[y])
 
     def _sweep_down(self, x, y):
+        if not self._entries:
+            return
         yield from self._entries[y][x:]
         while True:
             y = (y + 1) % len(self._entries)

eris/tools.py

@@ -106,8 +106,16 @@ def deps(**kwargs):
 
 def _call_in_process(module_name, func_name, *args, **kwargs):
     import importlib
-    fn = getattr(importlib.import_module(module_name), func_name)
-    return fn(*args, **kwargs)
+    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:
+            fn = getattr(importlib.import_module(module_name), func_name)
+            return fn(*args, **kwargs)
+        finally:
+            sys.stdout, sys.stderr = old_out, old_err
 
 
 def in_separate_process(func):

eris/worker.py

@@ -1,8 +1,6 @@
 
 import asyncio
-import contextlib
-import os
-import signal
+import concurrent.futures
 import sys
 
 import fill3
@@ -11,6 +9,23 @@ 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
+
+
 class Worker:
 
     AUTOSAVE_MESSAGE = "Auto-saving…"
@@ -19,33 +34,15 @@ class Worker:
     def __init__(self, compression):
         self.compression = compression
         self.result = None
-        self.process = None
-        self.child_pgid = None
-
-    async def create_process(self):
-        create = asyncio.create_subprocess_exec(
-            "eris-worker", 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"))
+        self.executor = concurrent.futures.ProcessPoolExecutor(max_workers=1)
 
     async def run_tool(self, path, tool):
-        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))
+        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)
 
     async def job_runner(self, screen, summary, log, jobs_added_event):
-        await self.create_process()
         while True:
             await jobs_added_event.wait()
             while True:
@@ -70,9 +67,9 @@ class Worker:
             jobs_added_event.clear()
 
     def kill(self):
-        if self.child_pgid is not None:
-            with contextlib.suppress(ProcessLookupError):
-                os.killpg(self.child_pgid, signal.SIGKILL)
+        for proc in self.executor._processes.values():
+            proc.kill()
+        self.executor.shutdown(wait=False, cancel_futures=True)
 
 
 def make_result_widget(text, result, compression):
@@ -83,23 +80,3 @@ 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()

pyproject.toml

@@ -25,7 +25,6 @@ dependencies = [
 
 [project.scripts]
 eris = "eris.__main__:entry_point"
-eris-worker = "eris.worker:main"
 eris-webserver = "eris.webserver:main"
 
 [project.urls]

tests/worker_test.py

@@ -19,16 +19,16 @@ class WorkerTestCase(unittest.TestCase):
         open("foo", "w").close()
 
     def tearDown(self):
+        if hasattr(self, "worker_"):
+            self.worker_.kill()
         shutil.rmtree(self.temp_dir)
         os.chdir(self.original_working_dir)
 
     def test_run_job(self):
         loop = asyncio.new_event_loop()
         compression = "none"
-        worker_ = worker.Worker(compression)
-        loop.run_until_complete(worker_.create_process())
-        worker_.process.stdin.write(f"{compression}\n".encode("utf-8"))
-        future = worker_.run_tool("foo", tools.metadata)
+        self.worker_ = worker.Worker(compression)
+        future = self.worker_.run_tool("foo", tools.metadata)
         status = loop.run_until_complete(future)
         self.assertEqual(status, tools.Status.ok)
         result_path = os.path.join(tools.CACHE_PATH, "foo-metadata")