Managing Hundreds of Coroutines: Production asyncio Task Lifecycles

At 02:13, a market-data gateway accepted 10,000 callbacks in two seconds. The application remained responsive, but most workers were stuck waiting for responses that would never arrive.

The process had not crashed. It had lost control of its own concurrency.

One timeout triggered an unbounded retry loop. Several tasks waited on the same failing dependency. Others continued after the supervisor believed they had stopped. When the process finally exited, the shutdown handler found hundreds of tasks that had neither completed nor acknowledged cancellation.

This failure mode is common in asyncio systems: creating tasks is easy, but managing their full lifecycle requires explicit policies for admission, monitoring, failure isolation, cancellation, and shutdown.

This article develops a production architecture for launching, supervising, and stopping hundreds or thousands of asynchronous operations. It explains where asyncio.gather remains useful, where asyncio.TaskGroup provides stronger lifecycle guarantees, and how signal handlers should coordinate graceful termination.

1. The Real Problem Is Task Ownership

Most asyncio code begins with a function and ends with a call to asyncio.run(). Production systems are more complicated because every task enters an implicit lifecycle:

1. The task is created.
2. The runtime schedules it.
3. The task waits on I/O.
4. The task succeeds, fails, or becomes unresponsive.
5. The supervisor observes the outcome.
6. The supervisor decides whether to retry, isolate, or cancel.
7. The application waits for cleanup before exiting.

A bare task call handles only the third and fourth states.

async def fetch_position(symbol: str) -> dict:
    response = await client.get(f"/positions/{symbol}")
    return response.json()


asyncio.run(fetch_position("AAPL.US"))

In a larger service, hundreds of calls to asyncio.create_task() produce hundreds of independently scheduled tasks. If nothing tracks their completion, no structure distinguishes expected work from abandoned work.

A production lifecycle must define measurable limits

Concern	Example limit
Maximum concurrent operations	250
Maximum queued operations	500
Individual request timeout	10 seconds
Stale-worker threshold	5 seconds
Maximum reconnect delay	30 seconds
Graceful shutdown window	15 seconds
Job-level retry attempts	3
Fatal dependency failures	Cancel the task group

The precise values depend on the workload. The important point is that lifecycle policy should exist before the first traffic spike.

Avoid creating one task per queued item

A queue with 10,000 pending items does not require 10,000 active tasks. It requires a bounded set of workers consuming those items.

                         ┌──────────────────────────┐
10,000 jobs ──────────► │ bounded queue: 500 items │
                         └─────────────┬────────────┘
                                       │
                         ┌─────────────▼────────────┐
                         │ 250 worker tasks        │
                         │ timeout + retry policy  │
                         └─────────────┬────────────┘
                                       │
                  ┌────────────────────▼────────────────────┐
                  │ bounded I/O concurrency: 250 requests    │
                  └────────────────────┬────────────────────┘
                                       │
                       ┌───────────────▼───────────────┐
                       │ dependency timeout / failure  │
                       └────────────────────────────────┘

This model provides backpressure. When consumers cannot keep up, the producer waits instead of consuming unbounded memory.

2. gather and TaskGroup Solve Different Problems

Python 3.11 introduced asyncio.TaskGroup, but it does not replace asyncio.gather. Their behavior differs in failure management, cancellation, and result handling.

Requirement	asyncio.gather	asyncio.TaskGroup
Start multiple awaitables	Yes	Yes, through child tasks
Return ordered results	Yes	No direct result collection
Wait for all children	Yes	Yes
One failure cancels siblings	No	Yes
Structured exception reporting	No	ExceptionGroup
Named child tasks	Limited	Yes
Clear task ownership	Weaker	Stronger

gather is useful for parallel result collection

asyncio.gather works well when the caller needs a deterministic list of results.

async def collect_health() -> list[dict]:
    responses = await asyncio.gather(
        check_rest_api(),
        check_websocket(),
        check_database(),
    )
    return responses

If one operation fails, gather propagates that exception to the caller. However, the remaining operations are not automatically cancelled. They may continue running in the background unless the caller cancels the gather future or provides explicit cleanup.

The return_exceptions=True option changes the behavior again:

results = await asyncio.gather(
    check_rest_api(),
    check_websocket(),
    check_database(),
    return_exceptions=True,
)

for result in results:
    if isinstance(result, Exception):
        log_failure(result)

This pattern is useful for best-effort probes, but it can hide a fatal infrastructure failure. A failed database check should not necessarily appear equivalent to a degraded optional cache.

TaskGroup provides structured failure propagation

A task group gives every child task an explicit owner and a shared failure boundary.

async def run_checks() -> None:
    async with asyncio.TaskGroup() as group:
        group.create_task(check_rest_api(), name="rest-health")
        group.create_task(check_websocket(), name="ws-health")
        group.create_task(check_database(), name="database-health")

If one child raises an unhandled exception, the task group:

1. Cancels the remaining children.
2. Waits for their cancellation handlers to complete.
3. Raises an ExceptionGroup containing the failures.

This behavior is usually what a supervisor wants. If a core dependency becomes unusable, continuing to schedule requests against it increases load without producing useful results.

Select exceptions deliberately

A task group treats ordinary unhandled exceptions as fatal to the group. Catch exceptions when the child should continue operating.

async def process_job(job_id: int) -> None:
    try:
        await execute_job(job_id)
    except RecoverableJobError:
        record_failure(job_id)
        return

If the exception should stop the entire group, allow it to propagate.

Do not use broad except Exception blocks as a substitute for lifecycle design. A broad catch may preserve a broken worker while hiding the condition that the supervisor should act on.

3. Production Architecture: Admission, Supervision, Termination

A resilient asyncio service separates four responsibilities:

1. Admission control limits concurrency and queue capacity.
2. Execution isolation contains job-level failures.
3. Supervision detects stalls and infrastructure failures.
4. Termination coordinates cancellation and graceful shutdown.

Architecture diagram

┌─────────────────────────────────────────────────────────────┐
│                       Producer / API                        │
│  validates jobs, applies admission control, limits queue    │
└─────────────────────────────┬───────────────────────────────┘
                              │ bounded queue
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                      TaskGroup supervisor                   │
│  owns workers + heartbeat monitor + lifecycle metrics       │
└───────────────┬──────────────────────────┬──────────────────┘
                │                          │
                ▼                          ▼
┌───────────────────────────┐  ┌──────────────────────────────┐
│ Worker pool               │  │ Heartbeat monitor            │
│ fixed concurrency         │  │ reports stale workers        │
│ job-level error isolation │  │ requests shutdown if needed  │
└─────────────┬─────────────┘  └──────────────────────────────┘
              │
              ▼
┌─────────────────────────────────────────────────────────────┐
│ I/O adapter                                                 │
│ timeout, heartbeat, reconnect, jitter, rate-limit handling │
└─────────────────────────────────────────────────────────────┘

The supervisor owns the task group. Workers do not own the application. This ownership model prevents a common mistake in which each worker creates additional untracked tasks.

4. A Complete Bounded Worker Pool

The following implementation creates a bounded queue, runs a fixed number of workers, isolates job failures, records heartbeats, and handles SIGINT and SIGTERM.

It requires Python 3.11 or later.

# lifecycle.py
from __future__ import annotations

import asyncio
import os
import random
import signal
import time
from dataclasses import dataclass
from typing import Optional


@dataclass
class Stats:
    completed: int = 0
    failed: int = 0
    cancelled: int = 0
    active: int = 0


@dataclass
class WorkerState:
    name: str
    last_heartbeat: float
    processed: int = 0


class Application:
    def __init__(
        self,
        *,
        total_jobs: int = 10_000,
        worker_count: int = 250,
        queue_capacity: int = 500,
        heartbeat_interval: float = 2.0,
        stale_after: float = 5.0,
    ) -> None:
        self.total_jobs = total_jobs
        self.worker_count = worker_count
        self.queue: asyncio.Queue[Optional[int]] = asyncio.Queue(
            maxsize=queue_capacity
        )
        self.heartbeat_interval = heartbeat_interval
        self.stale_after = stale_after

        self.stats = Stats()
        self.workers_finished = 0
        self.all_workers_done = asyncio.Event()

        self.worker_states = [
            WorkerState(
                name=f"worker-{index:03d}",
                last_heartbeat=time.monotonic(),
            )
            for index in range(worker_count)
        ]

    async def produce(self) -> None:
        """Fill the bounded queue and then insert one stop token per worker."""
        for job_id in range(self.total_jobs):
            await self.queue.put(job_id)

        for _ in range(self.worker_count):
            await self.queue.put(None)

    async def process_job(self, job_id: int) -> None:
        """Replace this method with a bounded I/O operation."""
        await asyncio.sleep(random.uniform(0.005, 0.080))

        # Simulate a recoverable job-level failure.
        if random.random() < 0.02:
            raise RuntimeError(f"job {job_id} failed")

    async def worker(self, index: int) -> None:
        state = self.worker_states[index]

        try:
            while True:
                job_id = await self.queue.get()

                if job_id is None:
                    self.queue.task_done()
                    return

                self.stats.active += 1
                state.last_heartbeat = time.monotonic()

                try:
                    await self.process_job(job_id)
                    self.stats.completed += 1
                    state.processed += 1
                except asyncio.CancelledError:
                    self.stats.cancelled += 1
                    state.last_heartbeat = time.monotonic()
                    raise
                except Exception as exc:
                    # Isolated job failures do not terminate the worker.
                    self.stats.failed += 1
                    state.last_heartbeat = time.monotonic()
                    log_job_failure(job_id, exc)
                finally:
                    self.stats.active -= 1
                    state.last_heartbeat = time.monotonic()
                    self.queue.task_done()
        finally:
            self.workers_finished += 1
            if self.workers_finished == self.worker_count:
                self.all_workers_done.set()

    async def monitor(self) -> None:
        while not self.all_workers_done.is_set():
            await asyncio.sleep(self.heartbeat_interval)

            now = time.monotonic()
            stale_workers = [
                state.name
                for state in self.worker_states
                if now - state.last_heartbeat > self.stale_after
            ]

            queue_ratio = self.queue.qsize() / self.queue.maxsize

            print(
                f"active={self.stats.active:>4} "
                f"completed={self.stats.completed:>6} "
                f"failed={self.stats.failed:>4} "
                f"queue={queue_ratio:>5.1%} "
                f"stale_workers={len(stale_workers)}"
            )

    def print_summary(self) -> None:
        print(
            "\nLifecycle summary\n"
            f"  completed: {self.stats.completed}\n"
            f"  failed:    {self.stats.failed}\n"
            f"  cancelled: {self.stats.cancelled}\n"
            f"  active:    {self.stats.active}\n"
        )


def log_job_failure(job_id: int, exc: Exception) -> None:
    print(f"job={job_id} failed: {exc}", flush=True)


async def main() -> None:
    application = Application(
        total_jobs=int(os.getenv("TOTAL_JOBS", "10000")),
        worker_count=int(os.getenv("WORKER_COUNT", "250")),
    )

    cancel_scope = None

    def request_shutdown() -> None:
        print("\nShutdown requested. Cancelling the task group...", flush=True)
        if cancel_scope is not None:
            cancel_scope.cancel()

    loop = asyncio.get_running_loop()
    installed_signals: list[signal.Signals] = []

    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, request_shutdown)
            installed_signals.append(sig)
        except NotImplementedError:
            # Windows does not implement loop signal handlers.
            print(f"Signal handler unavailable for {sig.name}", flush=True)

    try:
        async with asyncio.TaskGroup() as group:
            cancel_scope = group.cancel_scope

            group.create_task(application.produce(), name="producer")
            group.create_task(application.monitor(), name="heartbeat-monitor")

            for index in range(application.worker_count):
                group.create_task(
                    application.worker(index),
                    name=application.worker_states[index].name,
                )
    except asyncio.CancelledError:
        # Expected when SIGINT or SIGTERM cancels the task group's scope.
        pass
    finally:
        for sig in installed_signals:
            loop.remove_signal_handler(sig)

        application.print_summary()


if __name__ == "__main__":
    asyncio.run(main())

Run the example with:

python lifecycle.py

To test shutdown behavior, press Ctrl+C while the workers are active.

Why the queue capacity matters

The queue provides admission control. Without maxsize, a producer can enqueue work faster than dependencies can process it.

A bounded queue converts excess load into producer backpressure:

await self.queue.put(job_id)

This await point blocks the producer when the queue reaches capacity. That is preferable to allocating one task per pending callback because task objects also consume memory and scheduler capacity.

Why job exceptions stay inside the worker

The worker catches ordinary job exceptions because a failed position lookup should not stop unrelated position updates. Cancellation remains special:

except asyncio.CancelledError:
    self.stats.cancelled += 1
    raise

The exception must be re-raised. Swallowing it would cause the worker to continue running after the supervisor has requested shutdown.

5. Signals Should Cancel a Supervisor, Not Individual Tasks

A signal handler should not attempt to clean up every child independently. The supervisor owns the task group, so the signal handler should cancel the group's scope.

The key operation is:

cancel_scope.cancel()

This triggers cancellation throughout the task group. The group then waits for each child to acknowledge cancellation before asyncio.run() returns.

A handler that merely sets a flag is insufficient unless every worker polls that flag:

shutdown_requested = asyncio.Event()

Event-based shutdown works, but it requires a worker to reach an await boundary before it can react. Cancelling the task-group scope provides deterministic propagation.

Graceful and immediate shutdown are different modes

Not every cancellation should be immediate. A production system may use two levels:

Mode	Behavior	Use case
Graceful	Stop admitting work, drain the queue, then cancel	Planned maintenance
Immediate	Cancel active tasks and wait for cleanup	Process termination
Forced	Terminate the process after a grace deadline	Unresponsive application

A common graceful-shutdown sequence is:

1. Stop the HTTP or message listener.
2. Signal producers to stop.
3. Allow workers to drain the bounded queue.
4. Wait for in-flight work.
5. Cancel workers that exceed the deadline.
6. Close transport clients.

For example:

async def shutdown(grace_period: float = 15.0) -> None:
    application.accepting_work = False

    try:
        await asyncio.wait_for(
            application.queue.join(),
            timeout=grace_period,
        )
    except asyncio.TimeoutError:
        print("Grace period expired; cancelling workers")
        application.cancel_scope.cancel()

The grace-period decision belongs to the supervisor. Workers should not independently decide whether the application is shutting down.

6. Connecting a Resilient WebSocket Feed

Task management becomes more important when each job depends on a long-lived network connection. A resilient feed adapter should own its heartbeat, reconnect policy, and rate-limit behavior.

The following example uses environment-variable authentication and the WebSocket query-parameter authentication expected by TickDB.

# resilient_feed.py
from __future__ import annotations

import asyncio
import json
import os
import random
from urllib.parse import quote

import websockets
from websockets.exceptions import ConnectionClosed


class RateLimited(Exception):
    def __init__(self, retry_after: float) -> None:
        super().__init__(f"Rate limited; retry after {retry_after} seconds")
        self.retry_after = retry_after


class MarketFeed:
    def __init__(self) -> None:
        base_url = os.environ["TICKDB_WS_URL"].rstrip("?&")
        api_key = os.environ["TICKDB_API_KEY"]

        separator = "&" if "?" in base_url else "?"
        self.url = f"{base_url}{separator}api_key={quote(api_key, safe='')}"

        self.base_delay = 1.0
        self.max_delay = 30.0

    async def heartbeat(self, socket: websockets.WebSocketClientProtocol) -> None:
        while True:
            await asyncio.sleep(15)
            await socket.send(json.dumps({"cmd": "ping"}))

    async def consume(self, socket: websockets.WebSocketClientProtocol) -> None:
        while True:
            async with asyncio.timeout(30):
                raw_message = await socket.recv()

            try:
                message = json.loads(raw_message)
            except json.JSONDecodeError:
                print("Ignoring non-JSON WebSocket message")
                continue

            code = message.get("code")
            if code == 3001:
                retry_after = float(message.get("retry_after", 5))
                raise RateLimited(retry_after)

            if code in (1001, 1002):
                raise RuntimeError("Invalid or missing API key")

            if code == 2002:
                raise RuntimeError("Symbol not found")

            print(message)

    async def run(self) -> None:
        retry = 0

        while True:
            try:
                async with websockets.connect(
                    self.url,
                    ping_interval=20,
                    ping_timeout=10,
                    close_timeout=5,
                    max_queue=64,
                ) as socket:
                    retry = 0

                    async with asyncio.TaskGroup() as group:
                        group.create_task(
                            self.heartbeat(socket),
                            name="application-heartbeat",
                        )
                        group.create_task(
                            self.consume(socket),
                            name="message-consumer",
                        )
            except RateLimited as exc:
                print(exc)
                await asyncio.sleep(exc.retry_after)
            except (OSError, TimeoutError, ConnectionClosed) as exc:
                delay = min(
                    self.base_delay * (2 ** retry),
                    self.max_delay,
                )
                jitter = random.uniform(0, delay * 0.1)
                retry += 1

                print(
                    f"Connection failed: {exc}. "
                    f"Reconnecting in {delay + jitter:.2f}s"
                )
                await asyncio.sleep(delay + jitter)


if __name__ == "__main__":
    asyncio.run(MarketFeed().run())

What this adapter guarantees

The adapter provides five distinct protections:

1. Native protocol heartbeat: websockets sends protocol-level pings every 20 seconds.
2. Application heartbeat: the client sends {"cmd": "ping"} every 15 seconds.
3. Message timeout: a silent connection fails after 30 seconds.
4. Exponential reconnect delay: repeated failures increase the delay to a 30-second ceiling.
5. Jitter: reconnect attempts do not synchronize across hundreds of workers.

The jitter matters. Without it, a shared outage can cause every worker to reconnect at the same second, recreating the traffic spike that caused the outage.

Keep the connection inside the task group

The heartbeat and message consumer belong to the same task group. If the consumer encounters a fatal error, the group cancels the heartbeat task and waits for cleanup before reconnecting.

Avoid this pattern:

# Incorrect: the heartbeat may outlive the broken connection.
heartbeat_task = asyncio.create_task(self.heartbeat(socket))

try:
    await self.consume(socket)
finally:
    heartbeat_task.cancel()

The task group expresses the same ownership relationship without manual task cleanup.

7. Choosing the Right Concurrency Mechanism

The appropriate mechanism depends on the operation.

Requirement	Recommended mechanism
Collect several independent results	asyncio.gather
Run a failure-bound group of child tasks	asyncio.TaskGroup
Bound concurrent operations	Worker pool or asyncio.Semaphore
Process a streaming workload	asyncio.Queue plus fixed workers
Apply a deadline to part of a coroutine	asyncio.timeout
Cancel on a deadline	asyncio.wait_for
Run blocking I/O briefly	asyncio.to_thread
Run CPU-intensive work	ProcessPoolExecutor
Coordinate termination	Signal handler plus task-group cancellation

Semaphores versus worker pools

A semaphore limits simultaneous entry into a section:

async with semaphore:
    await execute_operation()

This is convenient when each item requires a different set of asynchronous steps. A worker pool is often easier to observe because every worker has a stable identity, heartbeat, and processed-item count.

Use a worker pool when you need:

• Per-worker health metrics.
• Controlled queue capacity.
• Long-running worker lifecycles.
• Easy replacement of stalled workers.

Use a semaphore when the operation is short and the surrounding function already handles the task lifecycle.

Deadlines must cover more than one await

asyncio.timeout works as a context manager around a block:

async with asyncio.timeout(2.0):
    await stream.read()
    await validate_payload()

This is more useful than calculating a deadline manually before several dependent operations.

Cancellation does not stop every underlying operation

Cancelling a coroutine waiting on asyncio.to_thread() does not terminate the thread. The event loop stops waiting for it, but the thread may continue consuming CPU or holding a resource.

Use ProcessPoolExecutor for genuinely CPU-bound work, and design thread-backed operations to tolerate late completion.

8. Observability: Measure the Lifecycle

A task count is not a sufficient health metric. A service can have thousands of tasks while completing no useful work.

Track at least these indicators:

Metric	Interpretation
Active workers	Current concurrency inside the bounded section
Queue depth	Pending work and backpressure pressure
Queue saturation	queue.qsize() / queue.maxsize
Completed jobs	Throughput over time
Failed jobs	Recoverable operation failures
Cancelled jobs	Shutdown or supervisor intervention
Stale workers	Workers that stopped reporting progress
Retry count	Dependency instability
Reconnect count	Connection instability
Task-group failures	Fatal infrastructure failures

Use structured logs rather than unstructured print statements in production:

logger.info(
    "worker_completed",
    extra={
        "worker": state.name,
        "job_id": job_id,
        "duration_ms": elapsed_ms,
        "queue_depth": self.queue.qsize(),
    },
)

Heartbeat monitoring should detect a stalled worker, not merely prove that the event loop is running. A task can remain scheduled while waiting forever on an operation that has no timeout.

9. Deployment Configurations by Workload

The correct configuration depends on whether the workload is I/O-bound, bursty, or connection-oriented.

Segment	Worker count	Queue capacity	Timeout	Recommended structure
Individual developer	10–25	100–250	5–10s	Worker pool with a bounded queue
Small quant team	50–100	250–1,000	5–15s	Task group plus metrics and signal handling
Institutional service	200–500	1,000–10,000	2–10s	Task group, shared semaphore, circuit breakers
Connection-oriented system	Depends on upstream limit	Small queue	Heartbeat-specific	One task group per managed connection

Worker count should not be configured from CPU count alone. An asyncio worker waiting on network I/O consumes little CPU while the event loop can manage many such workers.

Measure upstream limits before increasing concurrency. A service that opens 500 simultaneous requests to an API with a 100-request limit will create throttling, retries, and more latency.

10. Production Checklist

Before deploying an asyncio task supervisor, verify the following:

• Every task has an explicit owner.
• The task group owns all long-lived child tasks.
• Job-level failures are isolated intentionally.
• Fatal dependency failures cancel sibling tasks.
• Queue capacity prevents unbounded admission.
• Every network operation has a timeout.
• Every connection has a heartbeat.
• Reconnects use exponential backoff and jitter.
• Rate-limit responses use the server-provided delay.
• SIGINT and SIGTERM trigger coordinated cancellation.
• Cancellation handlers perform required cleanup.
• The application waits for cleanup before exiting.
• Metrics expose queue pressure and stale workers.
• Blocking and CPU-bound work runs outside the event loop.
• The system has been tested with partial failure and forced shutdown.

Conclusion

At 02:13, the overloaded gateway failed because task creation had been mistaken for task management.

The durable pattern is straightforward: bound admission, assign ownership, monitor progress, isolate recoverable failures, propagate fatal failures, and cancel through a supervisor.

asyncio.gather remains appropriate when the goal is to collect results. asyncio.TaskGroup is the stronger choice when several tasks form one operational unit and must share a failure boundary.

Next Steps

If you are learning asyncio lifecycle patterns, start with the bounded worker-pool example, then add cancellation tests before deploying it against real dependencies.

If you want to test the patterns against market data, sign up at tickdb.ai, generate an API key, and set TICKDB_API_KEY in your environment.

If you need historical OHLCV data for a larger backtesting platform, contact [email protected] for institutional options.

If you use AI coding assistants, search for the tickdb-market-data SKILL in your AI tool’s marketplace.

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.