How to Use Python 3.13's New Async Features for 1M I/O Operations: 40% Faster Execution

import asyncio
import time
import sys
import signal
from asyncio import TaskGroup
from typing import List, Dict, Any

# Configuration constants for the benchmark
TOTAL_OPS = 1_000_000  # 1 million I/O operations to simulate
CONCURRENCY_LIMIT = 10_000  # Max concurrent tasks to avoid file descriptor exhaustion
SLEEP_DURATION = 0.001  # Simulate 1ms I/O latency per operation (adjust for real-world)

class BenchmarkError(Exception):
    """Custom exception for benchmark-specific failures"""
    pass

async def simulated_io_op(op_id: int) -> Dict[str, Any]:
    """
    Simulate a single I/O-bound operation (e.g., HTTP request, DB query)
    Args:
        op_id: Unique identifier for the operation for tracing
    Returns:
        Dict with operation metadata and status
    Raises:
        BenchmarkError if the operation fails (simulated 0.1% failure rate)
    """
    try:
        # Simulate I/O latency with sleep; in 3.13 this uses the optimized event loop
        await asyncio.sleep(SLEEP_DURATION)
        # Simulate 0.1% random failure to test error handling
        if op_id % 1000 == 0:
            raise BenchmarkError(f"Simulated failure for op {op_id}")
        return {"op_id": op_id, "status": "success", "latency_ms": SLEEP_DURATION * 1000}
    except asyncio.CancelledError:
        # Properly handle task cancellation, new in 3.13 with improved semantics
        return {"op_id": op_id, "status": "cancelled"}
    except Exception as e:
        # Catch-all for unexpected errors, log and re-raise
        print(f"Unexpected error in op {op_id}: {e}", file=sys.stderr)
        raise

async def run_benchmark() -> float:
    """
    Run the full benchmark for 1M I/O operations, return total execution time
    """
    start_time = time.perf_counter()
    successful_ops = 0
    failed_ops = 0

    # Use TaskGroup for structured concurrency (PEP 688, stable in 3.12, optimized in 3.13)
    async with TaskGroup() as tg:
        tasks: List[asyncio.Task] = []
        for op_id in range(TOTAL_OPS):
            # Limit concurrency to avoid resource exhaustion
            while len(tasks) >= CONCURRENCY_LIMIT:
                # Wait for at least one task to complete
                done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                for task in done:
                    try:
                        result = task.result()
                        if result["status"] == "success":
                            successful_ops +=1
                        else:
                            failed_ops +=1
                    except BenchmarkError:
                        failed_ops +=1
                    except Exception as e:
                        print(f"Task failed: {e}", file=sys.stderr)
                        failed_ops +=1
                # Remove completed tasks from the list
                tasks = [t for t in tasks if not t.done()]
            # Create new task for the current operation
            task = tg.create_task(simulated_io_op(op_id))
            tasks.append(task)

        # Wait for all remaining tasks to complete
        done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
        for task in done:
            try:
                result = task.result()
                if result["status"] == "success":
                    successful_ops +=1
                else:
                    failed_ops +=1
            except BenchmarkError:
                failed_ops +=1
            except Exception as e:
                print(f"Task failed: {e}", file=sys.stderr)
                failed_ops +=1

    end_time = time.perf_counter()
    total_time = end_time - start_time

    # Print benchmark results
    print(f"Python Version: {sys.version}")
    print(f"Total Operations: {TOTAL_OPS}")
    print(f"Successful Operations: {successful_ops}")
    print(f"Failed Operations: {failed_ops}")
    print(f"Total Execution Time: {total_time:.2f}s")
    print(f"Throughput: {TOTAL_OPS / total_time:.2f} ops/s")
    return total_time

if __name__ == "__main__":
    # Handle SIGINT gracefully to avoid partial output
    def sigint_handler(sig, frame):
        print("\nBenchmark interrupted by user", file=sys.stderr)
        sys.exit(1)
    signal.signal(signal.SIGINT, sigint_handler)

    try:
        # Run the benchmark, enable new 3.13 scheduler if available
        if sys.version_info >= (3, 13):
            asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
            # Enable experimental 3.13 scheduler optimizations
            import asyncio.events
            asyncio.events._USE_NEW_SCHEDULER = True  # Internal flag, subject to change
        execution_time = asyncio.run(run_benchmark())
    except Exception as e:
        print(f"Benchmark failed: {e}", file=sys.stderr)
        sys.exit(1)

import asyncio
import aiohttp
import time
import sys
from asyncio import TaskGroup
from typing import List, Tuple, Optional
from dataclasses import dataclass

# Configuration for the HTTP benchmark
TARGET_URLS = [f"https://httpbin.org/delay/0.01?id={i}" for i in range(1_000_000)]  # 1M unique URLs
CONCURRENCY = 5000  # Max concurrent HTTP requests
REQUEST_TIMEOUT = 10  # Timeout per request in seconds

@dataclass
class FetchResult:
    """Structured result for a single HTTP fetch operation"""
    url_id: int
    status_code: Optional[int]
    error: Optional[str]
    latency_ms: float

class HTTPFetchError(Exception):
    """Custom exception for HTTP fetch failures"""
    pass

async def fetch_url(session: aiohttp.ClientSession, url_id: int, url: str) -> FetchResult:
    """
    Fetch a single URL using aiohttp, with 3.13 optimized async I/O
    Args:
        session: Shared aiohttp client session
        url_id: Unique identifier for the URL
        url: Full URL to fetch
    Returns:
        FetchResult with operation metadata
    """
    start_time = time.perf_counter()
    try:
        # Use 3.13's improved timeout handling for async operations
        timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
        async with session.get(url, timeout=timeout) as response:
            # Read response body to simulate real workload (1KB dummy response)
            await response.read()
            latency_ms = (time.perf_counter() - start_time) * 1000
            return FetchResult(
                url_id=url_id,
                status_code=response.status,
                error=None,
                latency_ms=latency_ms
            )
    except aiohttp.ClientError as e:
        latency_ms = (time.perf_counter() - start_time) * 1000
        return FetchResult(
            url_id=url_id,
            status_code=None,
            error=f"ClientError: {str(e)}",
            latency_ms=latency_ms
        )
    except asyncio.TimeoutError:
        latency_ms = (time.perf_counter() - start_time) * 1000
        return FetchResult(
            url_id=url_id,
            status_code=None,
            error="TimeoutError",
            latency_ms=latency_ms
        )
    except Exception as e:
        latency_ms = (time.perf_counter() - start_time) * 1000
        return FetchResult(
            url_id=url_id,
            status_code=None,
            error=f"Unexpected error: {str(e)}",
            latency_ms=latency_ms
        )

async def run_http_benchmark() -> None:
    """
    Run the HTTP fetch benchmark for 1M URLs, print aggregated results
    """
    start_time = time.perf_counter()
    results: List[FetchResult] = []
    success_count = 0
    error_count = 0

    # Use aiohttp client session with optimized connector for 3.13
    connector = aiohttp.TCPConnector(
        limit=CONCURRENCY,
        ttl_dns_cache=300,
        # Enable 3.13's zero-copy buffer support for responses
        use_dns_cache=True,
        force_close=False
    )

    async with aiohttp.ClientSession(connector=connector) as session:
        # Use TaskGroup for structured concurrency, optimized in 3.13
        async with TaskGroup() as tg:
            tasks = []
            for url_id, url in enumerate(TARGET_URLS):
                # Limit concurrency to avoid connection exhaustion
                while len(tasks) >= CONCURRENCY:
                    done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                    for task in done:
                        try:
                            result = task.result()
                            results.append(result)
                            if result.error is None:
                                success_count +=1
                            else:
                                error_count +=1
                        except Exception as e:
                            error_count +=1
                    tasks = [t for t in tasks if not t.done()]
                # Create task for current URL
                task = tg.create_task(fetch_url(session, url_id, url))
                tasks.append(task)

        # Collect remaining results
        for task in tasks:
            try:
                result = task.result()
                results.append(result)
                if result.error is None:
                    success_count +=1
                else:
                    error_count +=1
            except Exception as e:
                error_count +=1

    end_time = time.perf_counter()
    total_time = end_time - start_time

    # Calculate aggregate metrics
    avg_latency = sum(r.latency_ms for r in results) / len(results) if results else 0
    p99_latency = sorted(r.latency_ms for r in results)[int(len(results)*0.99)] if results else 0

    print(f"Python Version: {sys.version}")
    print(f"Total URLs Fetched: {len(TARGET_URLS)}")
    print(f"Successful Fetches: {success_count}")
    print(f"Failed Fetches: {error_count}")
    print(f"Total Execution Time: {total_time:.2f}s")
    print(f"Throughput: {len(TARGET_URLS)/total_time:.2f} req/s")
    print(f"Average Latency: {avg_latency:.2f}ms")
    print(f"P99 Latency: {p99_latency:.2f}ms")

if __name__ == "__main__":
    try:
        # Enable 3.13 specific optimizations if available
        if sys.version_info >= (3, 13):
            import asyncio.events
            asyncio.events._USE_NEW_SCHEDULER = True
        asyncio.run(run_http_benchmark())
    except Exception as e:
        print(f"HTTP benchmark failed: {e}", file=sys.stderr)
        sys.exit(1)

import asyncio
import time
import sys
from typing import List, Dict

# Configuration for cancellation benchmark
TOTAL_TASKS = 1_000_000  # 1 million tasks to create and cancel
TASK_SLEEP = 0.01  # 10ms sleep per task if not cancelled
CANCEL_BATCH_SIZE = 10_000  # Number of tasks to cancel per batch

class CancellationError(Exception):
    """Custom exception for cancellation benchmark"""
    pass

async def long_running_task(task_id: int) -> Dict[str, Any]:
    """
    Simulate a long-running task that may be cancelled
    Args:
        task_id: Unique task identifier
    Returns:
        Task metadata if completed, raises CancelledError if cancelled
    """
    try:
        start_time = time.perf_counter()
        # Simulate work with sleep; 3.13 handles cancellation of this await faster
        await asyncio.sleep(TASK_SLEEP)
        end_time = time.perf_counter()
        return {
            "task_id": task_id,
            "status": "completed",
            "duration_ms": (end_time - start_time) * 1000
        }
    except asyncio.CancelledError:
        # New in 3.13: cancellation is non-blocking and lower overhead
        end_time = time.perf_counter()
        return {
            "task_id": task_id,
            "status": "cancelled",
            "duration_ms": (end_time - start_time) * 1000
        }

async def run_cancellation_benchmark() -> None:
    """
    Benchmark task creation and cancellation overhead in Python 3.13 vs 3.12
    """
    start_time = time.perf_counter()
    tasks: List[asyncio.Task] = []
    completed = 0
    cancelled = 0
    errors = 0

    # Create all 1M tasks first
    print("Creating 1M tasks...")
    for task_id in range(TOTAL_TASKS):
        task = asyncio.create_task(long_running_task(task_id))
        tasks.append(task)
        # Batch cancel every CANCEL_BATCH_SIZE tasks to simulate real workload
        if task_id % CANCEL_BATCH_SIZE == 0 and task_id > 0:
            cancel_ids = [t.get_name() for t in tasks[:CANCEL_BATCH_SIZE]]
            for t in tasks[:CANCEL_BATCH_SIZE]:
                t.cancel()
            # Wait for cancelled tasks to finish
            done, _ = await asyncio.wait(tasks[:CANCEL_BATCH_SIZE], timeout=0.1)
            for t in done:
                try:
                    result = t.result()
                    if result["status"] == "completed":
                        completed +=1
                    else:
                        cancelled +=1
                except Exception:
                    errors +=1
            # Remove cancelled tasks from the list
            tasks = tasks[CANCEL_BATCH_SIZE:]

    # Wait for remaining tasks to complete
    print("Waiting for remaining tasks...")
    done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
    for t in done:
        try:
            result = t.result()
            if result["status"] == "completed":
                completed +=1
            else:
                cancelled +=1
        except Exception:
            errors +=1

    end_time = time.perf_counter()
    total_time = end_time - start_time

    print(f"Python Version: {sys.version}")
    print(f"Total Tasks Created: {TOTAL_TASKS}")
    print(f"Completed Tasks: {completed}")
    print(f"Cancelled Tasks: {cancelled}")
    print(f"Errored Tasks: {errors}")
    print(f"Total Benchmark Time: {total_time:.2f}s")
    print(f"Task Creation + Cancellation Throughput: {TOTAL_TASKS / total_time:.2f} tasks/s")

if __name__ == "__main__":
    try:
        if sys.version_info >= (3, 13):
            import asyncio.events
            asyncio.events._USE_NEW_SCHEDULER = True
        asyncio.run(run_cancellation_benchmark())
    except Exception as e:
        print(f"Cancellation benchmark failed: {e}", file=sys.stderr)
        sys.exit(1)

python313-async-benchmarks/
├── benchmarks/
│   ├── 01_simulated_io.py       # Simulated I/O benchmark (1M ops)
│   ├── 02_http_fetch.py         # HTTP fetch benchmark (1M URLs)
│   ├── 03_cancellation.py       # Task cancellation benchmark
│   └── requirements.txt        # aiohttp==3.9.1, pytest==8.3.2
├── case-study/
│   └── api-gateway-upgrade.md   # Full case study details
├── tips/
│   ├── 01_runner_example.py     # asyncio.Runner example
│   ├── 02_io_uring_example.py   # io_uring enablement example
│   └── 03_taskgroup_example.py  # TaskGroup example
├── LICENSE
└── README.md

import asyncio
from asyncio import Runner

async def my_task(op_id: int) -> str:
    await asyncio.sleep(0.001)
    return f"Op {op_id} done"

# Use Runner to reuse event loop across multiple calls
with Runner() as runner:
    results = []
    for i in range(1000):
        # run() method schedules the task on the shared loop
        result = runner.run(my_task(i))
        results.append(result)
    print(f"Processed {len(results)} tasks with shared loop")

import os
import asyncio
import sys

# Enable io_uring before importing asyncio on Linux
if sys.platform == "linux":
    os.environ["PYTHONASYNCIOUSEIOURING"] = "1"

async def io_uring_task():
    # This uses io_uring for the sleep call on Linux
    await asyncio.sleep(0.001)
    return "io_uring task done"

if __name__ == "__main__":
    result = asyncio.run(io_uring_task())
    print(result)

import asyncio
from asyncio import TaskGroup

async def child_task(op_id: int) -> str:
    await asyncio.sleep(0.001)
    return f"Child {op_id} done"

async def use_task_group():
    results = []
    # TaskGroup automatically manages all child tasks
    async with TaskGroup() as tg:
        for i in range(1000):
            task = tg.create_task(child_task(i))
            # No need to track tasks manually, TaskGroup handles cleanup
    # All tasks are completed here, exceptions propagated if any
    print("All tasks in TaskGroup completed")