fix(process-pool): bound shutdown latency (#469)

Fixes #297

Author: dexhunter

openevolve/process_parallel.py

@@ -332,6 +332,61 @@ def _run_iteration_worker(
         return SerializableResult(error=str(e), iteration=iteration)
 
 
+def _wait_for_processes(processes: tuple[mp.Process, ...], timeout: float) -> list[mp.Process]:
+    """Wait for process handles to observe worker exits without blocking indefinitely."""
+    deadline = time.monotonic() + timeout
+    alive = list(processes)
+    while alive:
+        next_alive = []
+        for process in alive:
+            try:
+                process.join(timeout=0)
+                if process.is_alive():
+                    next_alive.append(process)
+            except (AssertionError, ValueError):
+                continue
+        alive = next_alive
+        remaining = deadline - time.monotonic()
+        if not alive or remaining <= 0:
+            break
+        time.sleep(min(0.001, remaining))
+    return alive
+
+
+def _terminate_process_pool(executor: ProcessPoolExecutor) -> None:
+    """Cancel queued work and ensure all process-pool workers have exited."""
+    # Python < 3.14 has no public force-shutdown API. Capture only this
+    # executor's workers before shutdown clears its private process mapping.
+    process_map = getattr(executor, "_processes", None) or {}
+    processes = tuple(process_map.copy().values())
+    terminate_workers = getattr(executor, "terminate_workers", None)
+
+    if callable(terminate_workers):
+        terminate_workers()
+    else:
+        executor.shutdown(wait=False, cancel_futures=True)
+        for process in processes:
+            try:
+                if process.is_alive():
+                    process.terminate()
+            except (ProcessLookupError, ValueError):
+                continue
+
+    surviving_processes = _wait_for_processes(processes, timeout=1.0)
+    for process in surviving_processes:
+        try:
+            process.kill()
+        except (ProcessLookupError, ValueError):
+            continue
+
+    surviving_processes = _wait_for_processes(tuple(surviving_processes), timeout=1.0)
+    if surviving_processes:
+        logger.warning(
+            "Process-pool workers did not exit: %s",
+            [process.pid for process in surviving_processes],
+        )
+
+
 class ProcessParallelController:
     """Controller for process-based parallel evolution"""
 
@@ -428,9 +483,10 @@ def stop(self) -> None:
         """Stop the process pool"""
         self.shutdown_event.set()
 
-        if self.executor:
-            self.executor.shutdown(wait=True)
-            self.executor = None
+        executor = self.executor
+        self.executor = None
+        if executor:
+            _terminate_process_pool(executor)
 
         logger.info("Stopped process pool")
 

tests/test_process_parallel.py

@@ -4,17 +4,26 @@
 
 import asyncio
 import os
+from pathlib import Path
 import tempfile
 import unittest
 from unittest.mock import Mock, patch, MagicMock
 import time
-from concurrent.futures import Future
+from concurrent.futures import Future, ProcessPoolExecutor
+
+
+def _slow_test_worker(marker_path: str) -> str:
+    Path(marker_path).write_text(str(os.getpid()))
+    time.sleep(5)
+    return "finished"
+
 
 # Set dummy API key for testing
 os.environ["OPENAI_API_KEY"] = "test"
 
 from openevolve.config import Config, DatabaseConfig, EvaluatorConfig, LLMConfig, PromptConfig
 from openevolve.database import Program, ProgramDatabase
+from openevolve import process_parallel as process_parallel_module
 from openevolve.process_parallel import ProcessParallelController, SerializableResult
 
 
@@ -86,6 +95,55 @@ def test_controller_start_stop(self):
         self.assertIsNone(controller.executor)
         self.assertTrue(controller.shutdown_event.is_set())
 
+    def test_controller_stop_terminates_running_workers(self):
+        """Stopping the controller does not wait for stuck process-pool work."""
+        controller = ProcessParallelController(self.config, self.eval_file, self.database)
+        executor = ProcessPoolExecutor(max_workers=1)
+        controller.executor = executor
+        marker_path = os.path.join(self.test_dir, "worker.pid")
+        future = executor.submit(_slow_test_worker, marker_path)
+
+        deadline = time.monotonic() + 5
+        while not os.path.exists(marker_path) and time.monotonic() < deadline:
+            time.sleep(0.01)
+        self.assertTrue(os.path.exists(marker_path))
+        worker_pid = int(Path(marker_path).read_text())
+
+        started = time.monotonic()
+        controller.stop()
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 1)
+        self.assertIsNone(controller.executor)
+        self.assertTrue(controller.shutdown_event.is_set())
+        deadline = time.monotonic() + 1
+        while not future.done() and time.monotonic() < deadline:
+            time.sleep(0.01)
+        self.assertTrue(future.done())
+        with self.assertRaises(ProcessLookupError):
+            os.kill(worker_pid, 0)
+
+        # Cleanup is idempotent after the executor reference is cleared.
+        controller.stop()
+
+    def test_process_pool_shutdown_escalates_to_kill(self):
+        """Workers still alive after terminate are killed before returning."""
+        process = Mock()
+        process.is_alive.return_value = True
+        executor = Mock(spec=["_processes", "shutdown"])
+        executor._processes = {123: process}
+
+        with patch.object(
+            process_parallel_module,
+            "_wait_for_processes",
+            side_effect=[[process], []],
+        ):
+            process_parallel_module._terminate_process_pool(executor)
+
+        executor.shutdown.assert_called_once_with(wait=False, cancel_futures=True)
+        process.terminate.assert_called_once_with()
+        process.kill.assert_called_once_with()
+
     def test_database_snapshot_creation(self):
         """Test creating database snapshot for workers"""
         controller = ProcessParallelController(self.config, self.eval_file, self.database)