Best Practices for AsyncFlow¶

AsyncFlow is built on top of Python’s asyncio, combining asynchronous execution and task dependency management with a simple API. This page outlines recommended practices when using AsyncFlow effectively in your projects.

Why Best Practices Matter¶

Async programming can easily lead to:

• Hard-to-debug concurrency bugs 🚨
• Deadlocks or race conditions 💥
• Inefficient task scheduling ⚠️

By following these best practices, you can:

• Make your workflows reliable 🏅
• Maximize concurrency 🔥
• Keep code maintainable 🔧

Structure Your Tasks Clearly¶

• Define small, composable tasks that each do one thing well.
• Prefer pure functions or side-effect-free coroutines as tasks.
• Use @flow.function_task or @flow.executable_task decorators consistently.

@flow.function_task
async def fetch_data(url):
    ...

Inspect Task and Block State¶

Every future returned by a task or block call exposes a .state attribute that tracks its lifecycle:

fut = my_task()
print(fut.state)        # "PENDING" — not yet submitted

await asyncio.sleep(0)
print(fut.state)        # "RUNNING" or "DONE" depending on backend speed

await fut
print(fut.state)        # "DONE", "FAILED", or "CANCELLED"

React to Task Completion Without Background Coroutines¶

A common mistake is spawning one background coroutine per task to monitor its completion:

# ❌ Anti-pattern: one coroutine per task
async def watch(fut):
    await fut
    handle(fut.result())

for fut in futs:
    asyncio.create_task(watch(fut))   # N coroutines for N tasks

Use add_done_callback instead — one callback function handles all completions, zero extra coroutines:

# ✅ Correct: zero coroutines, purely event-driven
def on_done(fut):
    if fut.cancelled():
        print(f"cancelled — state={fut.state}")
    elif fut.exception():
        print(f"failed: {fut.exception()} — state={fut.state}")
    else:
        print(f"done: {fut.result()} — state={fut.state}")

for fut in [task1(), task2(), task3()]:
    fut.add_done_callback(on_done)

Use Dependencies Correctly¶

Tasks can depend on the output of other tasks: - Make your dependency graph explicit by passing tasks as arguments. - Don’t block unnecessarily: let AsyncFlow schedule dependencies.

asyncflow = await WorkflowEngine.create(dry_run=True)

async def task_a():
    asyncio.sleep(2) # (1)!

async def task_a():
    asyncio.sleep(2) # (2)!

async def task_c(task_a_fut, task_b_fut):
    asyncio.sleep(2)


result = task_c(task_a(), task_b()) # (3)!

1. Task A will run asynchronously (independently)
2. Task B will run asynchronously (independently)
3. Task C will wait implicitly for other tasks

Await Only at the Top Level¶

• Inside your workflow logic, don’t await intermediate tasks.
• Let AsyncFlow build the graph; only await the final or root tasks you care about.
• Awaiting early forces serialization and kills concurrency.

await task_a()
await task_b()

result = await task_c(task_a(), task_b())

Graceful Shutdown¶

Always shut down the flow explicitly when finished: - Releases resources (e.g., thread pools, processes). - Ensures a clean exit.

At the end of your async main:

await flow.shutdown()

Logging & Debugging¶

Enable detailed logs to diagnose issues:

export RADICAL_LOG_LVL=DEBUG

Logs show task dependencies, execution order, errors.

Clean Shutdown¶

• Use try/finally in your async main to ensure flow.shutdown() is always called, even on exceptions.