fix(evaluator): stop timed-out evaluations from hanging the worker

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)

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"""