Skip to content

Naming and Tracking Tasks for Observability

A service is slowly running out of memory, and a debug endpoint helpfully dumps asyncio.all_tasks(): 41,000 lines of <Task pending name='Task-883412' coro=<_wrap() running at ...>>. Every line is technically correct and none of them answers the question — what are these tasks for, which code started them, and how long have they been waiting? Tasks are the unit of work on the event loop, yet by default they carry no purpose, no owner and no age. A handful of conventions change that: meaningful names chosen at creation, a registry that records when and why each long-lived task started, metrics aggregated by name, and — on Python 3.14 — call graphs that show which task is waiting on which. This guide sets those up so that the next task dump explains itself.

Prerequisites

A task dump before and after naming 2 columns contrasting default names, named and registered. A task dump before and after naming default names Task-883412 ... no purpose no owner no age cannot be aggregated named and registered webhook.deliver:9f2c kind before the colon owner tag age and origin counts per kind The dump is only as useful as the names chosen when tasks were created.

1. Name tasks with a stable, parseable convention

Every task creation API accepts a name: asyncio.create_task(coro, name=...), TaskGroup.create_task(coro, name=...), and task.set_name() for tasks you did not create, such as the main task. Choose a convention that separates a stable kind from a variable identifier, so names can be aggregated as well as read.

import asyncio


def task_name(kind: str, ident: object | None = None) -> str:
    """'kind' is low-cardinality and aggregatable; 'ident' pinpoints one instance."""
    return kind if ident is None else f"{kind}:{ident}"


async def fetch(what: str, order_id: int) -> str:
    await asyncio.sleep(0.01)
    return f"{what} for {order_id}"


async def handle_order(order_id: int) -> list[str]:
    async with asyncio.TaskGroup() as tg:
        inventory = tg.create_task(fetch("inventory", order_id),
                                   name=task_name("order.fetch_inventory", order_id))
        price = tg.create_task(fetch("price", order_id),
                               name=task_name("order.fetch_price", order_id))
    return [inventory.result(), price.result()]


async def main() -> None:
    asyncio.current_task().set_name("main")
    async with asyncio.TaskGroup() as tg:
        for order_id in (101, 102):
            tg.create_task(handle_order(order_id), name=task_name("order.handle", order_id))
        await asyncio.sleep(0)
        print(sorted(t.get_name() for t in asyncio.all_tasks()))


asyncio.run(main())

The dotted kind reads like a metric name and sorts related tasks together; the suffix after the colon identifies the instance. Keep secrets and unbounded user input out of names — they end up in logs, dumps and metric labels.

Verify: the printed list contains main, order.handle:101, order.handle:102 and the four order.fetch_* tasks, and no Task-N defaults.

2. Record purpose and age in a registry

Names answer "what". For long-lived and background tasks you also need "since when" and "started by whom". A small registry, filled by a helper that creates the task, records that metadata and forgets each task when it finishes — without holding tasks alive longer than they should live.

import asyncio
import time
import traceback
from dataclasses import dataclass, field


@dataclass
class TaskInfo:
    name: str
    started: float
    origin: str                          # file:line of the code that created the task
    tags: dict[str, str] = field(default_factory=dict)


class TaskRegistry:
    def __init__(self) -> None:
        self._tasks: dict[asyncio.Task, TaskInfo] = {}

    def spawn(self, coro, *, name: str, **tags: str) -> asyncio.Task:
        caller = traceback.extract_stack(limit=2)[0]
        task = asyncio.create_task(coro, name=name)
        self._tasks[task] = TaskInfo(name, time.monotonic(), f"{caller.filename}:{caller.lineno}", tags)
        task.add_done_callback(self._tasks.pop)            # forget on completion; also a strong ref
        return task

    def snapshot(self, older_than: float = 0.0) -> list[dict]:
        now = time.monotonic()
        rows = [
            {"name": info.name, "age_s": round(now - info.started, 1),
             "origin": info.origin, **info.tags}
            for info in self._tasks.values() if now - info.started >= older_than
        ]
        return sorted(rows, key=lambda r: -r["age_s"])


registry = TaskRegistry()


async def main() -> None:
    registry.spawn(asyncio.sleep(3600), name="cache.refresher", owner="catalog")
    registry.spawn(asyncio.sleep(0.01), name="audit.flush", owner="compliance")
    await asyncio.sleep(0.05)
    for row in registry.snapshot():
        print(row)                                           # only the refresher remains


asyncio.run(main())

Because the registry holds a reference until the done-callback runs, it doubles as the strong-reference set recommended in preventing task garbage collection with strong references. The origin field is what turns a leak investigation from a search into a lookup: it names the line that started the task.

Verify: the snapshot shows only cache.refresher with its owner tag and origin; audit.flush has already removed itself.

3. Export task counts and ages by kind

Individual tasks are for debugging; aggregates are for dashboards and alerts. Group live tasks by the kind part of their name, and export the count and the age of the oldest task per kind.

import asyncio
import collections
import time


def task_metrics(registry: TaskRegistry) -> dict[str, dict[str, float]]:
    now = time.monotonic()
    counts: collections.Counter[str] = collections.Counter()
    oldest: dict[str, float] = {}
    for info in registry._tasks.values():
        kind = info.name.split(":", 1)[0]
        counts[kind] += 1
        oldest[kind] = max(oldest.get(kind, 0.0), now - info.started)
    return {kind: {"count": counts[kind], "oldest_age_s": round(oldest[kind], 3)} for kind in counts}


async def unnamed_total() -> int:
    """Tasks that escaped the convention: a coverage metric for naming discipline."""
    return sum(1 for t in asyncio.all_tasks() if t.get_name().startswith("Task-"))


async def main() -> None:
    asyncio.current_task().set_name("main")
    for i in range(3):
        registry.spawn(asyncio.sleep(0.2), name=f"webhook.deliver:{i}")
    asyncio.create_task(asyncio.sleep(0.2))                  # an unnamed one slips in
    await asyncio.sleep(0.05)
    print(task_metrics(registry), "unnamed:", await unnamed_total())


asyncio.run(main())

Two alerts come out of this naturally: a kind whose count grows without bound (a leak, detailed in tracking task growth in long-running services), and a kind whose oldest task is far older than its expected lifetime (a hang). The unnamed-task count is a coverage metric — it should trend towards zero as code adopts the convention.

Verify: the metrics show webhook.deliver with a count of 3, and the unnamed count reports the one task created without a name.

Reading per-kind task metrics 4 bars comparing count steady, ages short with the others. Reading per-kind task metrics count steady, ages short healthy unnamed tasks rising convention slipping oldest age far above lifetime a hang count grows without bound a leak Bar length encodes urgency, not a measured quantity. Alert per kind: growth means a leak, age means a hang.

4. Name tasks you do not create directly

Frameworks and libraries create tasks too: the ASGI server's per-request task, asyncio.to_thread helpers, library background loops. You cannot pass a name at their creation, but you can name the current task from inside it, and you can install a task factory that assigns a name derived from the coroutine when none was given.

import asyncio


def naming_task_factory(loop, coro, **kwargs):
    task = asyncio.Task(coro, loop=loop, **kwargs)
    if task.get_name().startswith("Task-"):
        code = getattr(coro, "cr_code", None)
        qualname = code.co_qualname if code is not None else type(coro).__name__
        task.set_name(f"unnamed.{qualname}")                 # still better than Task-883412
    return task


async def endpoint(request_id: str) -> None:
    asyncio.current_task().set_name(f"http.request:{request_id}")   # inside framework-made tasks
    await asyncio.sleep(0)


async def background_poll() -> None:
    await asyncio.sleep(0.01)


async def main() -> None:
    asyncio.current_task().set_name("main")                  # created before the factory existed
    asyncio.get_running_loop().set_task_factory(naming_task_factory)
    asyncio.create_task(background_poll())
    await asyncio.create_task(endpoint("r-17"))
    print(sorted(t.get_name() for t in asyncio.all_tasks()))


asyncio.run(main())

The factory keeps explicitly chosen names and replaces only defaults. Naming the current task at the top of a request handler gives framework-created tasks a meaningful name for the duration of the request.

Verify: the list includes unnamed.background_poll instead of Task-N; the endpoint task finished before the listing, but logging asyncio.current_task().get_name() inside it shows http.request:r-17.

5. Show who is waiting on whom with call graphs (Python 3.14)

A flat list cannot show that order.fetch_price:101 is stuck because order.handle:101 is waiting on it inside a TaskGroup. Python 3.14 adds asyncio.capture_call_graph() and asyncio.print_call_graph(), which walk from a task up through the tasks awaiting it, with each one's coroutine call stack.

import asyncio


async def slow_leaf() -> None:
    await asyncio.sleep(3)


async def handler(n: int) -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slow_leaf(), name=f"order.fetch_inventory:{n}")
        tg.create_task(slow_leaf(), name=f"order.fetch_price:{n}")


async def main() -> None:
    asyncio.current_task().set_name("main")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(handler(101), name="order.handle:101")
        await asyncio.sleep(0.1)
        stuck = next(t for t in asyncio.all_tasks() if t.get_name() == "order.fetch_price:101")
        asyncio.print_call_graph(stuck)                     # prints to stdout
        for t in asyncio.all_tasks():
            if t is not asyncio.current_task():
                t.cancel()                                  # end the demo without waiting 3 s


asyncio.run(main())

The output starts at Task(name='order.fetch_price:101') with its call stack ending in sleep(), then shows it is awaited by order.handle:101 inside TaskGroup.__aexit__, which is awaited by main. Names chosen in step 1 are what make that chain readable. The same release adds python -m asyncio ps <pid> and python -m asyncio pstree <pid> for inspecting a running process from outside; they need permission to read the target's memory and an interpreter build that exposes its runtime section — on one distribution build we tested they reported Failed to find the PyRuntime section, so check your deployment image before relying on them in an incident, and keep py-spy as a fallback.

Verify: the printed graph shows three named tasks linked by "Awaited by", with no Task-N names in the chain.

A call graph read bottom-up 3 stacked layers from order.fetch_price:101 to main. A call graph read bottom-up order.fetch_price:101 stack ends in sleep() the stuck work order.handle:101 TaskGroup.__aexit__ awaits the leaf main async with TaskGroup awaits the handler Each layer is awaited by the one below it; names make the chain legible.

Verification

Tasks are observable when:

  • Names follow the convention: new tasks are named kind:ident, and the unnamed-task metric trends to zero.
  • Long-lived tasks are registered: background tasks carry start time, origin and owner, and disappear from the registry when done.
  • Dashboards show kinds, not instances: counts and oldest ages are exported per kind with bounded label cardinality.
  • Framework tasks get names: request tasks are renamed on entry, and a task factory replaces default names.
  • Hangs are explainable: a task dump or call graph shows which named task is waiting on which.

Pitfalls & edge cases

  • High-cardinality metric labels. Exporting full names like webhook.deliver:9f2c… as labels explodes the metrics backend. Use only the kind as a label.
  • Registries that keep tasks forever. Forgetting the done-callback turns the registry into a leak. Always remove on completion.
  • Capturing stacks on every spawn in hot paths. traceback.extract_stack() is not free. Record origins for long-lived and background tasks, not for every short per-request child.
  • Renaming shared framework tasks. Some servers reuse a task across keep-alive requests; rename at the start of each request, not once per connection.
  • Assuming introspection tools work everywhere. External task inspection depends on the interpreter build and on ptrace permissions in containers; verify in the image you deploy.

Frequently Asked Questions

How do I name an asyncio task?

Pass name= to asyncio.create_task or to TaskGroup.create_task when creating it, or call task.set_name later, including asyncio.current_task().set_name for the task you are running in. The name appears in the task's repr, in asyncio.all_tasks dumps, in error messages and in Python 3.14 call graphs.

What is a good naming convention for asyncio tasks?

Combine a stable, low-cardinality kind with a variable identifier, for example order.fetch_price:101. The kind can be aggregated into metrics and alerts, while the identifier pinpoints one instance in a dump. Avoid putting secrets or unbounded user input into names.

How can I see which asyncio task is waiting on another task?

On Python 3.14 and later, call asyncio.print_call_graph(task) or asyncio.capture_call_graph(task) to walk from a task through the tasks awaiting it, with coroutine call stacks. Earlier versions only offer task.get_stack for one task at a time, so meaningful task names are the main way to connect them.

Why do my asyncio tasks show names like Task-883412?

Tasks created without a name get an automatically numbered default. Name tasks at creation, rename framework-created tasks from inside them with asyncio.current_task().set_name, and optionally install a task factory with loop.set_task_factory that assigns a name derived from the coroutine when none was provided.