diff --git a/openevolve/evaluator.py b/openevolve/evaluator.py index b1142ece5..30cb3bda7 100644 --- a/openevolve/evaluator.py +++ b/openevolve/evaluator.py @@ -13,6 +13,7 @@ import time import traceback import uuid +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple, Union import traceback @@ -56,6 +57,13 @@ def __init__( # Create a task pool for parallel evaluation self.task_pool = TaskPool(max_concurrency=config.parallel_evaluations) + # Dedicated executor (not the loop's default) so a timed-out evaluation's + # orphaned thread can't block asyncio.run()'s shutdown_default_executor(). + self._executor = ThreadPoolExecutor( + max_workers=max(1, config.parallel_evaluations), + thread_name_prefix="openevolve-eval", + ) + # Set up evaluation function if file exists self._load_evaluation_function() @@ -348,7 +356,7 @@ async def _direct_evaluate( # Create a coroutine that runs the evaluation function in an executor async def run_evaluation(): loop = asyncio.get_event_loop() - return await loop.run_in_executor(None, self.evaluate_function, program_path) + return await loop.run_in_executor(self._executor, self.evaluate_function, program_path) # Run the evaluation with timeout - let exceptions bubble up for retry handling result = await asyncio.wait_for(run_evaluation(), timeout=self.config.timeout) @@ -393,7 +401,9 @@ async def _cascade_evaluate( async def run_stage1(): loop = asyncio.get_event_loop() - return await loop.run_in_executor(None, module.evaluate_stage1, program_path) + return await loop.run_in_executor( + self._executor, module.evaluate_stage1, program_path + ) stage1_result = await asyncio.wait_for(run_stage1(), timeout=self.config.timeout) stage1_eval_result = self._process_evaluation_result(stage1_result) @@ -434,7 +444,9 @@ async def run_stage1(): async def run_stage2(): loop = asyncio.get_event_loop() - return await loop.run_in_executor(None, module.evaluate_stage2, program_path) + return await loop.run_in_executor( + self._executor, module.evaluate_stage2, program_path + ) stage2_result = await asyncio.wait_for(run_stage2(), timeout=self.config.timeout) stage2_eval_result = self._process_evaluation_result(stage2_result) @@ -496,7 +508,9 @@ async def run_stage2(): async def run_stage3(): loop = asyncio.get_event_loop() - return await loop.run_in_executor(None, module.evaluate_stage3, program_path) + return await loop.run_in_executor( + self._executor, module.evaluate_stage3, program_path + ) stage3_result = await asyncio.wait_for(run_stage3(), timeout=self.config.timeout) stage3_eval_result = self._process_evaluation_result(stage3_result) diff --git a/tests/test_evaluator_timeout.py b/tests/test_evaluator_timeout.py index d9053e4a0..b544e4867 100644 --- a/tests/test_evaluator_timeout.py +++ b/tests/test_evaluator_timeout.py @@ -389,6 +389,23 @@ async def run_test(): asyncio.run(run_test()) + def test_timeout_does_not_hang_process_on_cleanup(self): + """Regression test for issue #399: a timed-out evaluation's orphaned + thread must not block asyncio.run() cleanup and hang the worker.""" + + async def run_test(): + evaluator = self._create_evaluator(timeout=3) + program_code = "# SLEEP_LONG\ndef test(): return 'long'" + result = await evaluator.evaluate_program(program_code, "test_no_hang") + self.assertTrue(result.get("timeout")) + + outer_start = time.time() + asyncio.run(run_test()) + outer_elapsed = time.time() - outer_start + + # SLEEP_LONG takes 8s; before the fix this blocked until it finished. + self.assertLess(outer_elapsed, 5) + class TestTimeoutIntegration(unittest.TestCase): """Integration tests for timeout functionality"""