From d9f7ca084769a2b72845f069fa141c13177a0800 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sat, 15 Aug 2026 15:11:12 +0200 Subject: [PATCH] Fix engine creation on a non-main-thread event loop WorkflowEngine.create() crashed when the event loop was running in a secondary thread: _setup_signal_handlers() calls loop.add_signal_handler() which raises RuntimeError ("set_wakeup_fd only works in main thread of the main interpreter") off the main thread, while only NotImplementedError was caught. Skip signal handler installation upfront when the engine is not created on the main thread, and warn once that the host process has to manage shutdown itself. Exception handling inside the registration loop is left unchanged, so genuine errors still surface. There is no matching remove_signal_handler() site, so no teardown change is needed. Adds a unit test creating and shutting down an engine on a loop in a secondary thread. Co-Authored-By: Claude Fable 5 --- src/radical/asyncflow/workflow_manager.py | 14 ++++++++++- tests/unit/test_termination.py | 29 +++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/src/radical/asyncflow/workflow_manager.py b/src/radical/asyncflow/workflow_manager.py index da29a21..5c4acaf 100644 --- a/src/radical/asyncflow/workflow_manager.py +++ b/src/radical/asyncflow/workflow_manager.py @@ -7,6 +7,7 @@ import os import shlex import signal +import threading import time import uuid from collections import defaultdict, deque @@ -309,7 +310,18 @@ async def workflow_scope( def _setup_signal_handlers(self): """Register signal handlers for graceful shutdown on SIGHUP, SIGTERM, and - SIGINT.""" + SIGINT. + + Signal handlers can only be installed on a loop running in the main thread; + elsewhere registration is skipped and the engine runs without them. + """ + if threading.current_thread() is not threading.main_thread(): + logger.warning( + "running on a non-main-thread event loop; signal handlers not " + "installed - host process must manage shutdown" + ) + return + signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT) for sig in signals: try: diff --git a/tests/unit/test_termination.py b/tests/unit/test_termination.py index 96e1c11..583dbab 100644 --- a/tests/unit/test_termination.py +++ b/tests/unit/test_termination.py @@ -1,4 +1,5 @@ import asyncio +import logging import signal import threading import time @@ -300,3 +301,31 @@ async def mock_run_task(): # Verify completion assert shutdown_completed.is_set() + + def test_engine_creation_on_non_main_thread_loop(self, tmp_path, caplog): + """Test that an engine can be created on a loop in a secondary thread.""" + # Signal handlers cannot be installed off the main thread - the engine is + # expected to skip them (with a warning) instead of raising + errors = [] + + async def create_and_shutdown(): + engine = await WorkflowEngine.create(dry_run=True, work_dir=str(tmp_path)) + await engine.shutdown() + + def run_in_thread(): + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(create_and_shutdown()) + except Exception as e: + errors.append(e) + finally: + loop.close() + + thread = threading.Thread(target=run_in_thread, daemon=True) + with caplog.at_level(logging.WARNING, logger="radical.asyncflow"): + thread.start() + thread.join(timeout=60) + + assert not thread.is_alive(), "Engine creation in thread did not complete" + assert not errors, f"Engine creation failed off the main thread: {errors[0]!r}" + assert "signal handlers not installed" in caplog.text