Finding Bottlenecks in Asyncio: A Practical Approach

A few weeks ago at PythonSkillset, we had a server processing hundreds of WebSocket connections. It was working fine during development, but in production the response times suddenly became unpredictable. The usual profiling tools weren't helping—they showed everything running smoothly, but users were experiencing delays.

That's when I realized: profiling async code is fundamentally different from profiling synchronous Python.

Why Regular Profiling Falls Short

Traditional profilers like cProfile work by intercepting function calls. But asyncio doesn't run your code in a straightforward sequence—it switches between tasks constantly. When the profiler sees your async function spending 5 seconds "running," it could actually be waiting on 50 different await points spread across multiple tasks.

The real cost isn't in the CPU time, it's in the wall clock time your coroutines spend suspended, waiting for I/O or other resources.

The Tool That Actually Works

Python's built-in asyncio module gives us asyncio.all_tasks() and asyncio.current_task(), but for real profiling we need asyncio.Task objects and the event loop's internal state.

Here's the approach that saved our production server:

import asyncio
import time
from collections import defaultdict

class AsyncProfiler:
    def __init__(self):
        self.task_timings = defaultdict(list)
        self._loop = asyncio.get_event_loop()

    async def track_task(self, coro, name=None):
        task = asyncio.create_task(coro)
        task_name = name or coro.__name__
        start = time.monotonic()

        try:
            result = await task
            elapsed = time.monotonic() - start
            self.task_timings[task_name].append(elapsed)
            return result
        except Exception as e:
            elapsed = time.monotonic() - start
            self.task_timings[task_name].append(('error', elapsed, str(e)))
            raise

    def report(self):
        print("=== Task Performance Report ===")
        for task_name, timings in self.task_timings.items():
            if isinstance(timings[0], tuple):  # error case
                print(f"{task_name}: FAILED after {timings[0][1]:.3f}s - {timings[0][2]}")
            else:
                avg = sum(timings) / len(timings)
                print(f"{task_name}: avg {avg:.3f}s over {len(timings)} runs")

This profiler wraps your coroutines and measures actual elapsed time—including all the awaiting and rescheduling that happens underneath.

But even this has a blind spot.

The Real Problem: Contended Event Loop

What if the bottleneck isn't in any single task, but in how tasks compete for the event loop? If you have 50 database-querying tasks, each one might only take 50ms, but when they all try to run, the event loop gets clogged.

Here's where asyncio.gather() reveals its hidden cost:

# Bad: All start at once, compete for resources
async def bad_pattern():
    results = await asyncio.gather(*[slow_query() for _ in range(50)])

# Better: Use a semaphore to control concurrency
async def better_pattern():
    sem = asyncio.Semaphore(10)  # Only 10 at a time
    async def limited_query():
        async with sem:
            return await slow_query()
    results = await asyncio.gather(*[limited_query() for _ in range(50)])

The difference can be dramatic. Our production server had a database pool with 20 connections, but our code was trying to use 50 concurrent queries. The result wasn't faster—it was slower, because connections were queuing and timing out.

Profiling the Event Loop Itself

To catch this, we need to look at the event loop's scheduling behavior:

import asyncio
import time

class EventLoopProfiler:
    def __init__(self):
        self._original_loop = asyncio.get_event_loop()
        self.schedule_times = []
        self._next_task_id = 0

    def start(self):
        self._original_loop.call_soon(self._profile_step)

    def _profile_step(self):
        now = time.monotonic()
        # Track how many tasks are ready vs waiting
        tasks = asyncio.all_tasks()
        ready = sum(1 for t in tasks if not t.done())
        waiting = sum(1 for t in tasks if t.done())
        self.schedule_times.append((now, ready, waiting))

        # Reschedule for next cycle
        self._original_loop.call_later(0.1, self._profile_step)

Run this alongside your application for 30 seconds, and you'll see patterns:

profiler = EventLoopProfiler()
profiler.start()

# ... run your async app for 30 seconds ...

# Analyze results
import statistics
timestamps, ready_counts, waiting_counts = zip(*profiler.schedule_times)
print(f"Average ready tasks: {statistics.mean(ready_counts):.1f}")
print(f"Max ready tasks: {max(ready_counts)}")
print(f"Average waiting: {statistics.mean(waiting_counts):.1f}")

If you see the "ready tasks" count constantly high, your event loop is overloaded. If "waiting" is high, your tasks are spending too much time blocked on I/O.

The Surprise We Found

Using this profiler, we discovered something weird: our database queries were actually fast (10-15ms), but the tasks were spending 200-300ms in "ready" state before being scheduled. The culprit? A synchronous logging call in our middleware that was blocking the event loop for 50ms every time it ran.

The fix was simple—move logging to a separate thread using loop.run_in_executor():

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)

async def log_safely(message):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(executor, print, message)
    # Or use a proper async logger like aiologger

That single change cut our average response time by 60%.

Your Profiling Toolkit

Here's what to keep in your toolbox:

1. Wall-clock wrappers around your coroutines to measure true elapsed time
2. Event loop sampling to detect scheduling bottlenecks
3. Concurrency limiting with semaphores to prevent thundering herd problems
4. Escape hatch for blocking code: always use run_in_executor for anything that might block

The beauty of asyncio is in its concurrency model—but that same concurrency makes it hard to see what's really happening. With these tools, you can finally see the invisible waiting that's slowing down your async applications.

Have you found a clever way to profile async code? I'd love to hear about it. Drop a comment below or share your approach at PythonSkillset.