From f64a706885f67824887cc7f40c8b721de1c2424c Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 13 Oct 2026 01:54:55 +0200 Subject: [PATCH 1/4] Hot path: debug-only type checks, lazy logging, direct component creation Per-task overhead outside the scheduler dominated throughput after #91: profiled at 8,192 workflows, ~45% of the time was typeguard checks on task_callbacks, plus eager log formatting, telemetry argument building with telemetry off, and a @wraps closure built for every task call. - task_callbacks is type-checked only in debug mode: RADICAL_ASYNCFLOW_DEBUG, else RADICAL_DEBUG (1/true/yes/on), read at import. The test suite enables it in tests/conftest.py. __init__ stays type-checked (once per engine). - Per-task log calls use lazy arguments or an isEnabledFor guard; the per-task state line and the per-batch "Submitting" line move from INFO to DEBUG. - Telemetry call sites in registration, dependency resolution and the run loop build their arguments only when telemetry is enabled. - A decorated call creates its component via _create_component instead of building and calling a @wraps wrapper per call; the async check runs once per decorated function. - init_default_logger sets the logger level to its lowest handler level instead of NOTSET, so records no handler emits are not built. - docs/best_practice.md documents debug mode and replaces the RADICAL_LOG_LVL hint (never read by asyncflow); CHANGELOG updated. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 ++ docs/best_practice.md | 25 +- src/radical/asyncflow/logging.py | 5 +- src/radical/asyncflow/workflow_manager.py | 249 ++++++++++-------- tests/conftest.py | 9 +- tests/unit/test_debug_mode.py | 62 +++++ .../unit/test_task_registration_component.py | 36 ++- 7 files changed, 284 insertions(+), 117 deletions(-) create mode 100644 tests/unit/test_debug_mode.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 737f746..228cd75 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. With the root logger (no `logger_name`), DEBUG + records of other libraries are dropped too unless a handler accepts them. + ## [0.5.1] - 2026-08-20 ### Fixed diff --git a/docs/best_practice.md b/docs/best_practice.md index 00a3390..a6020d4 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 methods (via +`typeguard`). They cost about a third of the 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..1830a8e 100644 --- a/src/radical/asyncflow/logging.py +++ b/src/radical/asyncflow/logging.py @@ -319,8 +319,9 @@ 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 + logger.setLevel(min(h.level for h in logger.handlers)) # Capture warnings logging.captureWarnings(True) diff --git a/src/radical/asyncflow/workflow_manager.py b/src/radical/asyncflow/workflow_manager.py index 9659a93..ecb49eb 100644 --- a/src/radical/asyncflow/workflow_manager.py +++ b/src/radical/asyncflow/workflow_manager.py @@ -38,6 +38,23 @@ logger = logging.getLogger(__name__) +def _debug_enabled(environ=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). +_hot_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,6 +589,7 @@ 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): @@ -591,8 +609,11 @@ def wrapped(*args, **kwargs): **task_description_call, } - registered_func = self._handle_flow_component_registration( + return self._create_component( func, + args, + kwargs, + is_async=is_async, is_service=service, comp_type=comp_type, task_type=task_type, @@ -601,8 +622,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 +638,102 @@ def wrapped(*args, **kwargs): return outer - def _handle_flow_component_registration( + def _create_component( self, func: Callable, + args: tuple, + kwargs: dict, + *, + is_async: bool, 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_async (bool): Whether ``func`` is a coroutine function; only + async functions are supported 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 + asyncio.Future: The future tracking the component's execution. - Note: - Only async functions are supported. The decorator handles awaiting - async functions - and collecting their return values appropriately. + Raises: + TypeError: If ``func`` is not async. """ + if not is_async: + raise TypeError( + f"Function {func.__name__} must be async. " + "For sync functions, wrap them in async def." + ) - @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 - ) - except Exception as e: - if not comp_fut.done(): - comp_fut.set_exception(e) + 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, + } - # 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." + 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" + ) + 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 +829,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 +870,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"], @@ -1056,6 +1077,8 @@ def _notify_dependents(self, comp_uid: str): self._dependency_count[dependent_uid] -= 1 if self._dependency_count[dependent_uid] == 0: self._ready_queue.append(dependent_uid) + if self._telemetry is None: + continue dep_comp = self.components.get(dependent_uid) if dep_comp and dep_comp["type"] == TASK: dep_desc = dep_comp["description"] @@ -1241,8 +1264,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 +1276,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 +1304,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 +1334,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 +1460,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 +1702,7 @@ def handle_task_cancellation(self, task: dict, task_fut: asyncio.Future): self._mark_done(task["uid"]) return result - @typeguard.typechecked + @_hot_typechecked def task_callbacks( self, task: Any, state: str, service_callback: Optional[Callable] = None ) -> None: @@ -1715,7 +1750,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 +1768,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..e4c527b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,12 @@ # tests/conftest.py -import shutil +import os + +# 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") + +import shutil # noqa: E402 from pathlib import Path diff --git a/tests/unit/test_debug_mode.py b/tests/unit/test_debug_mode.py new file mode 100644 index 0000000..14c503d --- /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 +from typeguard import TypeCheckError + +from radical.asyncflow import NoopExecutionBackend, WorkflowEngine +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 not _debug_enabled(): + pytest.skip("debug mode disabled in this environment") + engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) + try: + with pytest.raises(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_task_registration_component.py b/tests/unit/test_task_registration_component.py index f1242f2..0189d92 100644 --- a/tests/unit/test_task_registration_component.py +++ b/tests/unit/test_task_registration_component.py @@ -27,21 +27,47 @@ 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_async=True, + 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_create_component_rejects_sync_function(): + engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) + + def test_func(x): + return x + + with pytest.raises(TypeError, match="must be async"): + engine._create_component( + test_func, + (1,), + {}, + is_async=False, + is_service=False, + comp_type=TASK, + task_type=FUNCTION, + ) + await engine.shutdown() @pytest.mark.asyncio From 5976291e1099a134da843c4850a26e64d92467ae Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 13 Oct 2026 02:00:40 +0200 Subject: [PATCH 2/4] =?UTF-8?q?Hot=20path:=20address=20local=20review=20?= =?UTF-8?q?=E2=80=94=20async=20check=20in=20decorator,=20docs,=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The decorator raises "must be async" itself; _create_component only creates components. - init_default_logger: a handler at NOTSET maps the logger level to 1 (accept all) rather than 0, which means "inherit" on named loggers. - CHANGELOG states the root and named logger effects of the level change; docs name task_callbacks as the switched method. - Tests: the sync-function rejection goes through the decorator; the type-check test skips on the import-time state; conftest imports stay at the top. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 ++++-- docs/best_practice.md | 4 ++-- src/radical/asyncflow/logging.py | 5 ++-- src/radical/asyncflow/workflow_manager.py | 23 +++++++------------ tests/conftest.py | 5 ++-- tests/unit/test_debug_mode.py | 7 +++--- .../unit/test_task_registration_component.py | 22 ++++++++---------- 7 files changed, 33 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 228cd75..480bf64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,8 +18,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `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. With the root logger (no `logger_name`), DEBUG - records of other libraries are dropped too unless a handler accepts them. + before they are built. With the root logger (no `logger_name`), records of + other libraries below that level are dropped too, unless the library sets a + level on its own logger. With `logger_name`, records at the configured + level are now emitted (they were filtered by the root logger's level) and + propagate to root handlers. ## [0.5.1] - 2026-08-20 diff --git a/docs/best_practice.md b/docs/best_practice.md index a6020d4..19816f5 100644 --- a/docs/best_practice.md +++ b/docs/best_practice.md @@ -184,8 +184,8 @@ state change; INFO and above show engine lifecycle events and errors. ### Debug mode -Debug mode adds runtime type checks on the per-task engine methods (via -`typeguard`). They cost about a third of the per-task overhead, so they are +Debug mode adds runtime type checks on the per-task engine method +`task_callbacks` (via `typeguard`). They cost about a third of the per-task overhead, so they are off by default. Enable them with an environment variable set before `radical.asyncflow` is imported: diff --git a/src/radical/asyncflow/logging.py b/src/radical/asyncflow/logging.py index 1830a8e..e1b9035 100644 --- a/src/radical/asyncflow/logging.py +++ b/src/radical/asyncflow/logging.py @@ -320,8 +320,9 @@ def init_default_logger( logger.addHandler(struct_handler) # Logger level = lowest handler level: records below every handler's - # level are dropped before a LogRecord is built - logger.setLevel(min(h.level for h in logger.handlers)) + # 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 ecb49eb..bbdd226 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,7 +38,7 @@ logger = logging.getLogger(__name__) -def _debug_enabled(environ=os.environ) -> bool: +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`` @@ -609,11 +609,16 @@ def wrapped(*args, **kwargs): **task_description_call, } + if not is_async: + raise TypeError( + f"Function {func.__name__} must be async. " + "For sync functions, wrap them in async def." + ) + return self._create_component( func, args, kwargs, - is_async=is_async, is_service=service, comp_type=comp_type, task_type=task_type, @@ -644,7 +649,6 @@ def _create_component( args: tuple, kwargs: dict, *, - is_async: bool, is_service: bool, comp_type: str, task_type: Optional[str], @@ -662,8 +666,6 @@ def _create_component( func (Callable): The decorated async function args (tuple): Positional call arguments kwargs (dict): Keyword call arguments (``workflow_id`` is consumed) - is_async (bool): Whether ``func`` is a coroutine function; only - async functions are supported 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 @@ -675,16 +677,7 @@ def _create_component( Returns: asyncio.Future: The future tracking the component's execution. - - Raises: - TypeError: If ``func`` is not async. """ - if not is_async: - raise TypeError( - f"Function {func.__name__} must be async. " - "For sync functions, wrap them in async def." - ) - comp_fut = asyncio.Future() comp_fut.state = "PENDING" diff --git a/tests/conftest.py b/tests/conftest.py index e4c527b..d0c2abb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,14 +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") -import shutil # noqa: E402 -from pathlib import Path - 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 index 14c503d..51631ef 100644 --- a/tests/unit/test_debug_mode.py +++ b/tests/unit/test_debug_mode.py @@ -3,9 +3,10 @@ import logging import pytest +import typeguard from typeguard import TypeCheckError -from radical.asyncflow import NoopExecutionBackend, WorkflowEngine +from radical.asyncflow import NoopExecutionBackend, WorkflowEngine, workflow_manager from radical.asyncflow.logging import init_default_logger from radical.asyncflow.workflow_manager import _debug_enabled @@ -35,8 +36,8 @@ def test_debug_enabled(environ, expected): 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 not _debug_enabled(): - pytest.skip("debug mode disabled in this environment") + if workflow_manager._hot_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(TypeCheckError): diff --git a/tests/unit/test_task_registration_component.py b/tests/unit/test_task_registration_component.py index 0189d92..c13061e 100644 --- a/tests/unit/test_task_registration_component.py +++ b/tests/unit/test_task_registration_component.py @@ -37,7 +37,6 @@ async def test_func(x): test_func, (1,), {}, - is_async=True, is_service=False, comp_type=TASK, task_type=FUNCTION, @@ -51,23 +50,20 @@ async def test_func(x): @pytest.mark.asyncio -async def test_create_component_rejects_sync_function(): +async def test_decorated_sync_function_rejected_at_call(): engine = await WorkflowEngine.create(backend=NoopExecutionBackend()) def test_func(x): return x - with pytest.raises(TypeError, match="must be async"): - engine._create_component( - test_func, - (1,), - {}, - is_async=False, - is_service=False, - comp_type=TASK, - task_type=FUNCTION, - ) - await engine.shutdown() + 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 From 8a6e065c1a4990a508ddf1d046f3a8ecf4dbf9ad Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 13 Oct 2026 21:23:10 +0200 Subject: [PATCH 3/4] =?UTF-8?q?Hot=20path:=20address=20second=20local=20re?= =?UTF-8?q?view=20=E2=80=94=20naming,=20flatter=20guards,=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _debug_typechecked names when the checks apply (matches _debug_enabled). - _notify_dependents: telemetry check joins the existing condition. - The decorator's async check is the first statement of the call path. - Docs drop a measured figure that would go stale; CHANGELOG entry for the logger level is shorter; test comment and imports tidied. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 ++----- docs/best_practice.md | 6 +++--- src/radical/asyncflow/workflow_manager.py | 24 ++++++++++++----------- tests/unit/test_debug_mode.py | 5 ++--- tests/unit/test_future_state.py | 2 +- 5 files changed, 21 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 480bf64..28a7b0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,11 +18,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `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. With the root logger (no `logger_name`), records of - other libraries below that level are dropped too, unless the library sets a - level on its own logger. With `logger_name`, records at the configured - level are now emitted (they were filtered by the root logger's level) and - propagate to root handlers. + before they are built. Configuring the root logger, records of other + libraries below that level are dropped too. ## [0.5.1] - 2026-08-20 diff --git a/docs/best_practice.md b/docs/best_practice.md index 19816f5..06e50eb 100644 --- a/docs/best_practice.md +++ b/docs/best_practice.md @@ -185,9 +185,9 @@ 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 cost about a third of the per-task overhead, so they are -off by default. Enable them with an environment variable set before -`radical.asyncflow` is imported: +`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_ASYNCFLOW_DEBUG=1 diff --git a/src/radical/asyncflow/workflow_manager.py b/src/radical/asyncflow/workflow_manager.py index bbdd226..1a7aa29 100644 --- a/src/radical/asyncflow/workflow_manager.py +++ b/src/radical/asyncflow/workflow_manager.py @@ -52,7 +52,7 @@ def _debug_enabled(environ: Mapping[str, str] = os.environ) -> bool: # Runtime type checks on per-task methods only in debug mode (read at import). -_hot_typechecked = typeguard.typechecked if _debug_enabled() else (lambda f: f) +_debug_typechecked = typeguard.typechecked if _debug_enabled() else (lambda f: f) class WorkflowEngine: @@ -594,6 +594,12 @@ def actual_decorator(func: Callable) -> Callable: @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 {} @@ -609,12 +615,6 @@ def wrapped(*args, **kwargs): **task_description_call, } - if not is_async: - raise TypeError( - f"Function {func.__name__} must be async. " - "For sync functions, wrap them in async def." - ) - return self._create_component( func, args, @@ -1070,10 +1070,12 @@ def _notify_dependents(self, comp_uid: str): self._dependency_count[dependent_uid] -= 1 if self._dependency_count[dependent_uid] == 0: self._ready_queue.append(dependent_uid) - if self._telemetry is None: - continue 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", @@ -1695,7 +1697,7 @@ def handle_task_cancellation(self, task: dict, task_fut: asyncio.Future): self._mark_done(task["uid"]) return result - @_hot_typechecked + @_debug_typechecked def task_callbacks( self, task: Any, state: str, service_callback: Optional[Callable] = None ) -> None: diff --git a/tests/unit/test_debug_mode.py b/tests/unit/test_debug_mode.py index 51631ef..f0d7cbd 100644 --- a/tests/unit/test_debug_mode.py +++ b/tests/unit/test_debug_mode.py @@ -4,7 +4,6 @@ import pytest import typeguard -from typeguard import TypeCheckError from radical.asyncflow import NoopExecutionBackend, WorkflowEngine, workflow_manager from radical.asyncflow.logging import init_default_logger @@ -36,11 +35,11 @@ def test_debug_enabled(environ, expected): 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._hot_typechecked is not typeguard.typechecked: + 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(TypeCheckError): + with pytest.raises(typeguard.TypeCheckError): engine.task_callbacks({"uid": "task.000001"}, 42) finally: await engine.shutdown() 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"] From 850485b18599ca646752b2a896a06f19854aa3b8 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 13 Oct 2026 22:38:12 +0200 Subject: [PATCH 4/4] tests: docformatter rewrap in test_debug_mode Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_debug_mode.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_debug_mode.py b/tests/unit/test_debug_mode.py index f0d7cbd..05e4596 100644 --- a/tests/unit/test_debug_mode.py +++ b/tests/unit/test_debug_mode.py @@ -33,8 +33,8 @@ def test_debug_enabled(environ, 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.""" + """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())