diff --git a/CHANGELOG.md b/CHANGELOG.md index 737f746..28a7b0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Changed + +- **Per-task overhead** — runtime type checks on the per-task path + (`task_callbacks`) run only in debug mode (`RADICAL_ASYNCFLOW_DEBUG` or + `RADICAL_DEBUG`, see *Logging & Debugging* in the best-practice guide); + per-task log messages format lazily; telemetry event arguments are built + only when telemetry is enabled; a decorated call creates its component + directly instead of building a wrapper per call. +- **Log levels** — per-task messages (` is in state`, + `Submitting tasks/blocks`) are logged at DEBUG instead of INFO. +- **`init_default_logger`** sets the logger level to the lowest level of its + handlers instead of `NOTSET`, so records no handler would emit are dropped + before they are built. Configuring the root logger, records of other + libraries below that level are dropped too. + ## [0.5.1] - 2026-08-20 ### Fixed diff --git a/docs/best_practice.md b/docs/best_practice.md index 00a3390..06e50eb 100644 --- a/docs/best_practice.md +++ b/docs/best_practice.md @@ -172,11 +172,32 @@ await flow.shutdown() ## Logging & Debugging Enable detailed logs to diagnose issues: +```python +import logging +from radical.asyncflow.logging import init_default_logger + +init_default_logger(logging.DEBUG) +``` + +DEBUG logs show task registration, dependencies, submission and every task +state change; INFO and above show engine lifecycle events and errors. + +### Debug mode + +Debug mode adds runtime type checks on the per-task engine method +`task_callbacks` (via `typeguard`). They add noticeable per-task overhead, +so they are off by default. Enable them with an environment variable set +before `radical.asyncflow` is imported: + ```bash -export RADICAL_LOG_LVL=DEBUG +export RADICAL_ASYNCFLOW_DEBUG=1 ``` -Logs show task dependencies, execution order, errors. +`RADICAL_DEBUG` is honoured as well; when both are set, +`RADICAL_ASYNCFLOW_DEBUG` decides, so `RADICAL_ASYNCFLOW_DEBUG=0` disables +debug mode in an environment that sets `RADICAL_DEBUG`. The values `1`, +`true`, `yes` and `on` (any case) enable it; any other value disables it. +The test suite runs in debug mode. --- diff --git a/src/radical/asyncflow/logging.py b/src/radical/asyncflow/logging.py index 97b5fde..e1b9035 100644 --- a/src/radical/asyncflow/logging.py +++ b/src/radical/asyncflow/logging.py @@ -319,8 +319,10 @@ def init_default_logger( logger.addHandler(struct_handler) - # Set logger level - logger.setLevel(logging.NOTSET) + # Logger level = lowest handler level: records below every handler's + # level are dropped before a LogRecord is built. A handler at NOTSET (0) + # accepts everything; level 1 expresses that without meaning "inherit". + logger.setLevel(min(h.level for h in logger.handlers) or 1) # Capture warnings logging.captureWarnings(True) diff --git a/src/radical/asyncflow/workflow_manager.py b/src/radical/asyncflow/workflow_manager.py index 9659a93..1a7aa29 100644 --- a/src/radical/asyncflow/workflow_manager.py +++ b/src/radical/asyncflow/workflow_manager.py @@ -17,7 +17,7 @@ from functools import wraps from itertools import count from pathlib import Path -from typing import Any, AsyncIterator, Callable, Optional, Union +from typing import Any, AsyncIterator, Callable, Mapping, Optional, Union import typeguard @@ -38,6 +38,23 @@ logger = logging.getLogger(__name__) +def _debug_enabled(environ: Mapping[str, str] = os.environ) -> bool: + """Debug mode: ``RADICAL_ASYNCFLOW_DEBUG`` if set, else ``RADICAL_DEBUG``. + + The specific variable takes precedence, so ``RADICAL_ASYNCFLOW_DEBUG=0`` + disables debug mode even when ``RADICAL_DEBUG`` is set. Values ``1``, + ``true``, ``yes`` and ``on`` (any case) enable it; anything else disables it. + """ + for name in ("RADICAL_ASYNCFLOW_DEBUG", "RADICAL_DEBUG"): + if name in environ: + return environ[name].strip().lower() in ("1", "true", "yes", "on") + return False + + +# Runtime type checks on per-task methods only in debug mode (read at import). +_debug_typechecked = typeguard.typechecked if _debug_enabled() else (lambda f: f) + + class WorkflowEngine: """An asynchronous workflow manager that uses asyncio event loops and coroutines to manage and execute workflow components (blocks and/or tasks) within Directed Acyclic @@ -572,10 +589,17 @@ def actual_decorator(func: Callable) -> Callable: task_description_def = param.default setattr(func, "__task_description__", task_description_def) + is_async = inspect.iscoroutinefunction(func) @wraps(func) def wrapped(*args, **kwargs): try: + if not is_async: + raise TypeError( + f"Function {func.__name__} must be async. " + "For sync functions, wrap them in async def." + ) + # Extract task_description from call-time kwargs task_description_call = kwargs.pop("task_description", {}) or {} @@ -591,8 +615,10 @@ def wrapped(*args, **kwargs): **task_description_call, } - registered_func = self._handle_flow_component_registration( + return self._create_component( func, + args, + kwargs, is_service=service, comp_type=comp_type, task_type=task_type, @@ -601,8 +627,6 @@ def wrapped(*args, **kwargs): capture_stdio=capture_stdio, ) - return registered_func(*args, **kwargs) - except Exception as e: # Add context to any errors that occur # during registration/execution @@ -619,107 +643,90 @@ def wrapped(*args, **kwargs): return outer - def _handle_flow_component_registration( + def _create_component( self, func: Callable, + args: tuple, + kwargs: dict, + *, is_service: bool, comp_type: str, task_type: Optional[str], task_backend_specific_kwargs: Optional[dict] = None, target_backend: Optional[str] = None, capture_stdio: bool = False, - ): - """Handles registration of async tasks and blocks as workflow components. + ) -> asyncio.Future: + """Creates a workflow component for one call of a decorated function. - Creates a decorator that manages component registration for async functions. - The decorator creates appropriate asyncio.Future to track execution state. + Returns the component's future at once; registration runs in an + asyncio task, so the component is registered on the next loop + iteration (after awaiting ``func`` for executable and prompt tasks). Args: - func (Callable): Async function to be registered as a workflow component + func (Callable): The decorated async function + args (tuple): Positional call arguments + kwargs (dict): Keyword call arguments (``workflow_id`` is consumed) is_service (bool): Whether the component is a service comp_type (str): Component type (e.g., "task", "block") task_type (str): Task type, determines result handling task_backend_specific_kwargs (dict, optional): Backend-specific kwargs. - Defaults to None. + target_backend (str, optional): Name of the backend to run on. capture_stdio (bool): If True, redirect stdout/stderr to files in the backend's _work_dir. task["stdout"] and task["stderr"] will be file paths instead of decoded strings. Only applies to executable tasks. Returns: - Callable: A decorator that: - - Wraps the original async function - - Registers it as a workflow component - - Returns a future tracking the component's execution - - Note: - Only async functions are supported. The decorator handles awaiting - async functions - and collecting their return values appropriately. + asyncio.Future: The future tracking the component's execution. """ + comp_fut = asyncio.Future() + comp_fut.state = "PENDING" + + # Extract call-time workflow_id before storing kwargs or calling the function + explicit_workflow_id = kwargs.pop("workflow_id", None) + + comp_desc = { + "args": args, + "function": func, + "kwargs": kwargs, + "is_service": is_service, + "task_backend_specific_kwargs": task_backend_specific_kwargs or {}, + "target_backend": target_backend, + "capture_stdio": capture_stdio, + "_explicit_workflow_id": explicit_workflow_id, + } - @wraps(func) - def wrapper(*args, **kwargs): - # Create async future - we only support async - comp_fut = asyncio.Future() - comp_fut.state = "PENDING" - - # Extract call-time workflow_id before storing kwargs or calling the function - explicit_workflow_id = kwargs.pop("workflow_id", None) - - comp_desc = { - "args": args, - "function": func, - "kwargs": kwargs, - "is_service": is_service, - "task_backend_specific_kwargs": task_backend_specific_kwargs or {}, - "target_backend": target_backend, - "capture_stdio": capture_stdio, - "_explicit_workflow_id": explicit_workflow_id, - } - - # Only handle async functions - if asyncio.iscoroutinefunction(func): - - async def async_wrapper(): - try: - if task_type == EXECUTABLE: - cmd = await func(*args, **kwargs) - if not cmd or not isinstance(cmd, str): - raise ValueError( - f"Executable task '{func.__name__}' must return " - "a non-empty command string" - ) - parts = shlex.split(cmd) - if not parts: - raise ValueError( - f"Executable task '{func.__name__}' must return " - "a non-empty command string" - ) - comp_desc[EXECUTABLE] = parts[0] - comp_desc["arguments"] = parts[1:] - elif task_type == PROMPT: - comp_desc[PROMPT] = await func(*args, **kwargs) - else: - comp_desc[EXECUTABLE] = None - return self._register_component( - comp_fut, comp_type, comp_desc, task_type + async def register(): + try: + if task_type == EXECUTABLE: + cmd = await func(*args, **kwargs) + if not cmd or not isinstance(cmd, str): + raise ValueError( + f"Executable task '{func.__name__}' must return " + "a non-empty command string" ) - except Exception as e: - if not comp_fut.done(): - comp_fut.set_exception(e) - - # FIXME: assign name for this comp (comp uid) - asyncio.create_task(async_wrapper()) - else: - # Raise error for non-async functions since we only support async - raise TypeError( - f"Function {func.__name__} must be async. " - "For sync functions, wrap them in async def." + parts = shlex.split(cmd) + if not parts: + raise ValueError( + f"Executable task '{func.__name__}' must return " + "a non-empty command string" + ) + comp_desc[EXECUTABLE] = parts[0] + comp_desc["arguments"] = parts[1:] + elif task_type == PROMPT: + comp_desc[PROMPT] = await func(*args, **kwargs) + else: + comp_desc[EXECUTABLE] = None + return self._register_component( + comp_fut, comp_type, comp_desc, task_type ) + except Exception as e: + if not comp_fut.done(): + comp_fut.set_exception(e) - return comp_fut + # FIXME: assign name for this comp (comp uid) + asyncio.create_task(register()) - return wrapper + return comp_fut def _register_component( self, @@ -815,8 +822,10 @@ def _register_component( self.dependencies[comp_desc["uid"]] = comp_deps logger.debug( - f"Registered {comp_type}: '{comp_desc['name']}' " - f"with id of {comp_desc['uid']}" + "Registered %s: '%s' with id of %s", + comp_type, + comp_desc["name"], + comp_desc["uid"], ) # Update dependency tracking @@ -854,20 +863,25 @@ def _register_component( if comp_type != BLOCK: comp_fut.cancel = self._setup_future_cancel_hook(comp_fut, comp_desc["uid"]) - self._emit( - "TaskCreated", - task_id=comp_desc["uid"], - workflow_id=comp_desc.get("workflow_id"), - attributes={ - "executable": comp_desc["_tel_executable"], - "task_type": comp_desc["_tel_task_type"], - "depends_on": [d["uid"] for d in comp_deps], - }, - ) + if self._telemetry is not None: + self._emit( + "TaskCreated", + task_id=comp_desc["uid"], + workflow_id=comp_desc.get("workflow_id"), + attributes={ + "executable": comp_desc["_tel_executable"], + "task_type": comp_desc["_tel_task_type"], + "depends_on": [d["uid"] for d in comp_deps], + }, + ) # Emit TaskResolved immediately for tasks with no dependencies. # Tasks with deps will emit from _notify_dependents when the count hits 0. - if comp_type == TASK and self._dependency_count.get(comp_desc["uid"], -1) == 0: + if ( + self._telemetry is not None + and comp_type == TASK + and self._dependency_count.get(comp_desc["uid"], -1) == 0 + ): self._emit( "asyncflow.TaskResolved", task_id=comp_desc["uid"], @@ -1057,7 +1071,11 @@ def _notify_dependents(self, comp_uid: str): if self._dependency_count[dependent_uid] == 0: self._ready_queue.append(dependent_uid) dep_comp = self.components.get(dependent_uid) - if dep_comp and dep_comp["type"] == TASK: + if ( + self._telemetry is not None + and dep_comp + and dep_comp["type"] == TASK + ): dep_desc = dep_comp["description"] self._emit( "asyncflow.TaskResolved", @@ -1241,8 +1259,9 @@ async def run(self): "output_files" ): logger.debug( - f"Linking implicit file(s): from {dep_desc['name']} " - f"to {comp_desc['name']}" + "Linking implicit file(s): from %s to %s", + dep_desc["name"], + comp_desc["name"], ) self.backend.link_implicit_data_deps( dep_desc, comp_desc @@ -1252,9 +1271,10 @@ async def run(self): for output_file in dep_desc["metadata"]["output_files"]: if output_file in comp_desc["metadata"]["input_files"]: logger.debug( - f"Linking explicit file ({output_file}) " - f"from {dep_desc['name']} " - f"to {comp_desc['name']}" + "Linking explicit file (%s) from %s to %s", + output_file, + dep_desc["name"], + comp_desc["name"], ) data_dep = self.backend.link_explicit_data_deps( src_task=dep_desc, @@ -1279,8 +1299,9 @@ async def run(self): and input_basename not in dependency_output_files ): logger.debug( - f"Staging {input_file} " - f"to {comp_desc['name']} work dir" + "Staging %s to %s work dir", + input_file, + comp_desc["name"], ) data_dep = self.backend.link_explicit_data_deps( src_task=None, @@ -1308,12 +1329,17 @@ async def run(self): continue to_submit.append(comp_desc) - res_deps = [dep["name"] for dep in dependencies] - msg = f"Ready to submit: {comp_desc['name']}" - msg += f" with resolved dependencies: {res_deps}" - logger.debug(msg) + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Ready to submit: %s with resolved dependencies: %s", + comp_desc["name"], + [dep["name"] for dep in dependencies], + ) - if self.components[comp_uid]["type"] == TASK: + if ( + self._telemetry is not None + and self.components[comp_uid]["type"] == TASK + ): self._emit( "TaskQueued", task_id=comp_uid, @@ -1429,8 +1455,12 @@ async def submit(self, objects: list) -> None: tasks = [t for t in objects if t and BLOCK not in t["uid"]] blocks = [b for b in objects if b and TASK not in b["uid"]] - logger.info(f"Submitting {len(objects)} tasks/blocks for execution") - logger.debug(f"Submitting: {[b['name'] for b in objects]}") + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Submitting %d tasks/blocks for execution: %s", + len(objects), + [b["name"] for b in objects], + ) if tasks: if self._telemetry is not None: @@ -1667,7 +1697,7 @@ def handle_task_cancellation(self, task: dict, task_fut: asyncio.Future): self._mark_done(task["uid"]) return result - @typeguard.typechecked + @_debug_typechecked def task_callbacks( self, task: Any, state: str, service_callback: Optional[Callable] = None ) -> None: @@ -1715,7 +1745,7 @@ def wait_and_set(): state not in self.task_states_map.terminal_states and state != self.task_states_map.RUNNING ): - logger.debug(f"Non-relevant task state received: {state}. Skipping state.") + logger.debug("Non-relevant task state received: %s. Skipping state.", state) return task_obj = task @@ -1733,7 +1763,7 @@ def wait_and_set(): comp = self.components[task_dct["uid"]] task_fut = comp["future"] comp_desc = comp["description"] - logger.info(f"{task_dct['uid']} is in {state} state") + logger.debug("%s is in %s state", task_dct["uid"], state) if service_callback: # service tasks are marked done by a backend specific diff --git a/tests/conftest.py b/tests/conftest.py index 89caec4..d0c2abb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,13 @@ # tests/conftest.py +import os import shutil from pathlib import Path +# Debug mode (runtime type checks) for the test suite; read when +# radical.asyncflow is imported, so set it before any test module imports it. +# An explicit RADICAL_ASYNCFLOW_DEBUG in the environment wins. +os.environ.setdefault("RADICAL_ASYNCFLOW_DEBUG", "1") + def pytest_sessionfinish(session, exitstatus): root = Path(__file__).parent diff --git a/tests/unit/test_debug_mode.py b/tests/unit/test_debug_mode.py new file mode 100644 index 0000000..05e4596 --- /dev/null +++ b/tests/unit/test_debug_mode.py @@ -0,0 +1,62 @@ +"""Debug mode: runtime type checks on per-task methods, logger level.""" + +import logging + +import pytest +import typeguard + +from radical.asyncflow import NoopExecutionBackend, WorkflowEngine, workflow_manager +from radical.asyncflow.logging import init_default_logger +from radical.asyncflow.workflow_manager import _debug_enabled + + +@pytest.mark.parametrize( + "environ, expected", + [ + ({}, False), + ({"RADICAL_ASYNCFLOW_DEBUG": "1"}, True), + ({"RADICAL_ASYNCFLOW_DEBUG": "TRUE"}, True), + ({"RADICAL_ASYNCFLOW_DEBUG": "yes"}, True), + ({"RADICAL_ASYNCFLOW_DEBUG": "on"}, True), + ({"RADICAL_ASYNCFLOW_DEBUG": "0"}, False), + ({"RADICAL_ASYNCFLOW_DEBUG": ""}, False), + ({"RADICAL_DEBUG": "TRUE"}, True), + ({"RADICAL_DEBUG": "false"}, False), + # the specific variable decides when set + ({"RADICAL_ASYNCFLOW_DEBUG": "0", "RADICAL_DEBUG": "TRUE"}, False), + ({"RADICAL_ASYNCFLOW_DEBUG": "1", "RADICAL_DEBUG": "0"}, True), + ], +) +def test_debug_enabled(environ, expected): + assert _debug_enabled(environ) is expected + + +@pytest.mark.asyncio +async def test_task_callbacks_typechecked_in_debug_mode(): + """The suite runs in debug mode (tests/conftest.py): task_callbacks checks its + argument types.""" + if workflow_manager._debug_typechecked is not typeguard.typechecked: + pytest.skip("debug mode disabled when radical.asyncflow was imported") + engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) + try: + with pytest.raises(typeguard.TypeCheckError): + engine.task_callbacks({"uid": "task.000001"}, 42) + finally: + await engine.shutdown() + + +def test_logger_level_follows_handlers(): + """Records below every handler's level are dropped at the logger.""" + name = "radical.asyncflow.test_debug_mode" + logger = init_default_logger(logging.WARNING, logger_name=name, clear_handlers=True) + try: + assert logger.level == logging.WARNING + assert not logger.isEnabledFor(logging.DEBUG) + + # the lowest handler level wins + logger.addHandler(logging.NullHandler(level=logging.DEBUG)) + init_default_logger(logging.INFO, logger_name=name) + assert logger.level == logging.DEBUG + finally: + logger.handlers.clear() + logger.setLevel(logging.NOTSET) diff --git a/tests/unit/test_future_state.py b/tests/unit/test_future_state.py index ede4980..35b6851 100644 --- a/tests/unit/test_future_state.py +++ b/tests/unit/test_future_state.py @@ -53,7 +53,7 @@ async def some_task(): return "ok" some_task() - await asyncio.sleep(0.05) # let async_wrapper register the component + await asyncio.sleep(0.05) # let the component register (asyncio task) comp = next(iter(engine.components.values())) task_fut = comp["future"] diff --git a/tests/unit/test_task_registration_component.py b/tests/unit/test_task_registration_component.py index f1242f2..c13061e 100644 --- a/tests/unit/test_task_registration_component.py +++ b/tests/unit/test_task_registration_component.py @@ -27,21 +27,43 @@ async def dummy_task(x): @pytest.mark.asyncio -async def test_handle_flow_component_registration_registers_function(): +async def test_create_component_registers_function(): engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) async def test_func(x): return x - decorated = engine._handle_flow_component_registration( - func=test_func, + fut = engine._create_component( + test_func, + (1,), + {}, + is_service=False, comp_type=TASK, task_type=FUNCTION, - is_service=False, task_backend_specific_kwargs={}, ) - assert callable(decorated) + assert isinstance(fut, asyncio.Future) + await fut + assert fut.task["function"] is test_func + await engine.shutdown() + + +@pytest.mark.asyncio +async def test_decorated_sync_function_rejected_at_call(): + engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) + + def test_func(x): + return x + + try: + decorated = engine.function_task(test_func) + with pytest.raises( + TypeError, match="Error in decorated function 'test_func'.*must be async" + ): + decorated(1) + finally: + await engine.shutdown() @pytest.mark.asyncio