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