diff --git a/openevolve/process_parallel.py b/openevolve/process_parallel.py index 2d1660c75..b2cfeab78 100644 --- a/openevolve/process_parallel.py +++ b/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") diff --git a/tests/test_process_parallel.py b/tests/test_process_parallel.py index 733f51109..23c6f9219 100644 --- a/tests/test_process_parallel.py +++ b/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)