Deep Dive into Python Concurrency: Asyncio Event Loop, Scheduling, and High‑Performance Web Apps

Asyncio Deep Dive: Event Loop and Coroutine Scheduling

The article begins with a simplified event‑loop implementation that demonstrates how tasks are queued, scheduled, and executed. The SimpleEventLoop class maintains a ready‑task list and a scheduled‑task list, uses selectors.DefaultSelector for I/O monitoring, and repeatedly pops tasks from the ready queue to drive them forward until they raise StopIteration, at which point the result is printed.

import asyncio
import selectors

class SimpleEventLoop:
    """Simplified event‑loop example"""
    def __init__(self):
        self._ready = []          # ready task queue
        self._scheduled = []      # timed tasks
        self._selector = selectors.DefaultSelector()

    def create_task(self, coro):
        """Wrap a coroutine into a Task"""
        task = asyncio.Task(coro)
        self._ready.append(task)
        return task

    def run_until_complete(self, coro):
        """Run until the given coroutine finishes"""
        task = self.create_task(coro)
        while self._ready or self._scheduled:
            # Execute all ready tasks
            while self._ready:
                current = self._ready.pop(0)
                try:
                    # Advance one step of the coroutine
                    result = current.send(None)
                except StopIteration as e:
                    # Coroutine finished
                    print(f'Task completed with result: {e.value}')
                    # In a real asyncio loop, I/O events would be handled here

Coroutine State Transitions

A small TaskState enum (PENDING, RUNNING, DONE) is defined, followed by an async function that prints messages before and after await points, illustrating how a coroutine suspends and resumes. The function creates a task, prints its internal state, and returns a result.

from enum import Enum
import asyncio

class TaskState(Enum):
    PENDING = 1
    RUNNING = 2
    DONE = 3

async def demonstrate_scheduling():
    """Show coroutine scheduling and state changes"""
    print("1. Coroutine starts")
    await asyncio.sleep(1)               # suspension point, control returns to loop
    print("2. Resumed and continues")
    future = asyncio.Future()
    await future                         # second suspension point
    print("3. Coroutine finishes")
    return "result"

# Create task and display its state
task = asyncio.create_task(demonstrate_scheduling())
print(f"Task state: {task._state}")

Synchronous vs Asynchronous Web‑Request Performance

A PerformanceTester class is introduced to benchmark async (aiohttp) against multithreaded sync (requests + ThreadPoolExecutor) under 100 concurrent requests to http://httpbin.org/delay/1. The class defines async_test, sync_threaded_test, and run_comparison methods, prints elapsed times, request counts, and the speed‑up factor, and visualizes the results with Matplotlib.

import asyncio
import aiohttp
import requests
import threading
from concurrent.futures import ThreadPoolExecutor
import time
from statistics import mean
import matplotlib.pyplot as plt

class PerformanceTester:
    """Performance comparison test framework"""
    def __init__(self, url, concurrent_requests=100):
        self.url = url
        self.concurrent = concurrent_requests

    async def async_test(self):
        """Asynchronous test"""
        async with aiohttp.ClientSession() as session:
            tasks = []
            start = time.time()
            for _ in range(self.concurrent):
                task = session.get(self.url)
                tasks.append(task)
            responses = await asyncio.gather(*tasks)
            elapsed = time.time() - start
            return elapsed, len(responses)

    def sync_threaded_test(self):
        """Multithreaded synchronous test"""
        def make_request():
            return requests.get(self.url)
        with ThreadPoolExecutor(max_workers=50) as executor:
            start = time.time()
            results = list(executor.map(make_request, range(self.concurrent)))
            elapsed = time.time() - start
            return elapsed, len(results)

    def run_comparison(self):
        """Run the benchmark and print results"""
        print(f"Testing {self.concurrent} concurrent requests to {self.url}")
        loop = asyncio.get_event_loop()
        async_time, async_count = loop.run_until_complete(self.async_test())
        sync_time, sync_count = self.sync_threaded_test()
        print("=" * 50)
        print(f"Asyncio result: {async_time:.2f}s, processed {async_count} requests")
        print(f"Multithreaded sync result: {sync_time:.2f}s, processed {sync_count} requests")
        print(f"Performance gain: {sync_time/async_time:.1f}x")
        self.visualize_results(async_time, sync_time)

    def visualize_results(self, async_time, sync_time):
        """Result visualization"""
        labels = ['asyncio async', 'multithreaded sync']
        times = [async_time, sync_time]
        plt.figure(figsize=(8, 5))
        bars = plt.bar(labels, times, color=['skyblue', 'lightcoral'])
        plt.ylabel('Processing time (seconds)')
        plt.title(f'Concurrent request performance comparison ({self.concurrent} requests)')
        for bar, time_val in zip(bars, times):
            plt.text(bar.get_x() + bar.get_width()/2, bar.get_height(), f'{time_val:.2f}s', ha='center', va='bottom')
        plt.tight_layout()
        plt.show()

if __name__ == '__main__':
    tester = PerformanceTester('http://httpbin.org/delay/1', concurrent_requests=100)
    tester.run_comparison()

The benchmark shows that under low concurrency (<100 connections) the difference is minor, but from 100‑1000 connections the async version begins to outperform, and beyond 1000 connections async is markedly faster while using far less memory.

Memory usage comparison: each thread in the sync model consumes ~8 MiB of stack memory (≈8 GiB for 1000 threads), whereas each coroutine consumes ~1 KiB (≈1 MiB for 1000 coroutines).

Practical Async Web Service Architecture

An end‑to‑end async web service built with aiohttp, asyncpg, and aioredis is presented. The service sets up routes, middleware that measures request latency, and handlers that demonstrate cache‑first reads, database queries, batch inserts, and parallel calls to multiple external sources.

from aiohttp import web
import asyncpg
import aioredis
from datetime import datetime

class AsyncWebService:
    """High‑performance async web service example"""
    def __init__(self):
        self.app = web.Application()
        self.setup_routes()
        self.setup_middleware()

    async def init_db(self):
        """Initialize database connection pool"""
        self.db_pool = await asyncpg.create_pool(
            user='user', password='password', database='database', host='localhost',
            min_size=5, max_size=20)

    async def init_cache(self):
        """Initialize Redis connection pool"""
        self.redis = await aioredis.create_redis_pool(
            'redis://localhost', minsize=5, maxsize=20)

    def setup_middleware(self):
        """Configure middleware"""
        @web.middleware
        async def timing_middleware(request, handler):
            start = datetime.now()
            response = await handler(request)
            elapsed = (datetime.now() - start).total_seconds()
            response.headers['X-Response-Time'] = f'{elapsed:.3f}s'
            return response
        self.app.middlewares.append(timing_middleware)

    def setup_routes(self):
        """Define routes"""
        self.app.router.add_get('/api/users/{id}', self.get_user)
        self.app.router.add_post('/api/users', self.create_user)
        self.app.router.add_get('/api/products', self.list_products)

    async def get_user(self, request):
        """Get user info with cache"""
        user_id = request.match_info['id']
        cached = await self.redis.get(f'user:{user_id}')
        if cached:
            return web.json_response({'cached': True, 'data': cached})
        async with self.db_pool.acquire() as conn:
            user = await conn.fetchrow('SELECT * FROM users WHERE id = $1', user_id)
            if user:
                await self.redis.setex(f'user:{user_id}', 300, user['name'])
                return web.json_response(dict(user))
            return web.json_response({'error': 'User not found'}, status=404)

    async def create_user(self, request):
        """Create users with async batch processing"""
        data = await request.json()
        tasks = []
        async with self.db_pool.acquire() as conn:
            async with conn.transaction():
                for user_data in data['users']:
                    task = conn.execute(
                        '''INSERT INTO users (name, email) VALUES ($1, $2)''',
                        user_data['name'], user_data['email'])
                    tasks.append(task)
            await asyncio.gather(*tasks)
        return web.json_response({'status': 'success', 'created': len(data['users'])})

    async def list_products(self, request):
        """Product list aggregating multiple sources"""
        tasks = [self.fetch_products_from_source('source1'),
                 self.fetch_products_from_source('source2'),
                 self.fetch_products_from_cache()]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        products = []
        for result in results:
            if isinstance(result, Exception):
                print(f"Query failed: {result}")
            elif result:
                products.extend(result)
        return web.json_response({'count': len(products), 'products': products})

    async def fetch_products_from_source(self, source):
        """Simulate external API call"""
        await asyncio.sleep(0.1)
        return [{'id': 1, 'name': f'Product from {source}'}]

    async def fetch_products_from_cache(self):
        """Placeholder for cache retrieval"""
        return []

    async def startup(self, app):
        """Initialize resources on startup"""
        await self.init_db()
        await self.init_cache()
        print("Application startup complete")

    async def cleanup(self, app):
        """Clean up resources on shutdown"""
        await self.db_pool.close()
        self.redis.close()
        await self.redis.wait_closed()
        print("Application cleanup complete")

    def run(self):
        """Run the web service"""
        self.app.on_startup.append(self.startup)
        self.app.on_cleanup.append(self.cleanup)
        web.run_app(self.app, host='0.0.0.0', port=8080, access_log=None)

if __name__ == '__main__':
    service = AsyncWebService()
    service.run()

Best Practices and Optimizations

The article shows how to fine‑tune the database connection pool (min/max size, max queries, idle timeout, command timeout) and presents an async batch‑processing helper that processes items in configurable chunks while yielding briefly to avoid resource exhaustion.

# Optimized DB pool configuration
async def get_optimized_db_pool():
    return await asyncpg.create_pool(
        dsn='postgresql://user:pass@localhost/db',
        min_size=5,
        max_size=50,          # adjust based on CPU cores
        max_queries=50000,    # reuse connections
        max_inactive_connection_lifetime=300,
        command_timeout=60)

# Async batch processing pattern
async def batch_process(items, batch_size=100):
    """Asynchronous batch processing"""
    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        tasks = [process_item(item) for item in batch]
        batch_results = await asyncio.gather(*tasks)
        results.extend(batch_results)
        await asyncio.sleep(0.001)  # tiny pause to prevent exhaustion
    return results

Decision Matrix for Choosing a Concurrency Model

CPU‑intensive workloads : Use multiprocessing to bypass the GIL and achieve true parallelism.

I/O‑bound with < 1000 concurrent connections : Use multithreading for simplicity and low development cost.

High‑concurrency I/O (>1000 connections) : Choose asyncio coroutines for massive concurrency with minimal memory footprint.

Mixed tasks (CPU + I/O) : Adopt a hybrid architecture—processes for CPU work, coroutines for I/O.

Conclusion

The article equips readers with a thorough understanding of Python's async primitives, concrete performance data comparing async and sync approaches, a production‑ready async web service skeleton, and clear guidelines for selecting the most suitable concurrency strategy based on workload characteristics.