From 4baf37c50574eccd7e4753758c4991c7c8fda730 Mon Sep 17 00:00:00 2001 From: ghosts6 Date: Tue, 25 Aug 2026 20:17:46 -0400 Subject: [PATCH] fix(evaluator): stop timed-out evaluations from hanging the worker asyncio.wait_for(loop.run_in_executor(None, ...)) only stops waiting on timeout; the blocking evaluate/evaluate_stageN call keeps running in the asyncio loop's default executor thread, since Python cannot force-kill a running thread. process_parallel.py calls asyncio.run() once per iteration, and asyncio.run()'s cleanup blocks in shutdown_default_executor() until every thread in that default executor finishes - so one timed-out evaluation (e.g. an evolved program with an infinite loop) hangs the entire worker process, matching issue #399. Give Evaluator its own dedicated ThreadPoolExecutor and pass it explicitly to every run_in_executor call instead of None. It is never the loop's default executor, so asyncio.run() no longer waits on it and a timed-out call's orphaned thread can't block cleanup. Resolves #399 --- openevolve/evaluator.py | 22 ++++++++++++++++++---- tests/test_evaluator_timeout.py | 17 +++++++++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/openevolve/evaluator.py b/openevolve/evaluator.py index b1142ece50..30cb3bda70 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 d9053e4a07..b544e48676 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"""