Async Programming

Async Programming Skill

File Organization

• SKILL.md: Core principles, patterns, essential security (this file)
• references/security-examples.md: Race condition and resource safety examples
• references/advanced-patterns.md: Advanced async patterns and optimization

Validation Gates

Gate 0.1: Domain Expertise Validation

• Status: PASSED
• Expertise Areas: asyncio, Tokio, race conditions, resource management, concurrent safety

Gate 0.2: Vulnerability Research

• Status: PASSED (3+ issues for MEDIUM-RISK)
• Research Date: 2025-11-20
• Issues: CVE-2024-12254 (asyncio memory), Redis race condition (CVE-2023-28858/9)

Gate 0.11: File Organization Decision

• Decision: Split structure (MEDIUM-RISK, ~400 lines main + references)

---

1. Overview

Risk Level: MEDIUM

Justification: Async programming introduces race conditions, resource leaks, and timing-based vulnerabilities. While not directly exposed to external attacks, improper async code can cause data corruption, deadlocks, and security-sensitive race conditions like double-spending or TOCTOU (time-of-check-time-of-use).

You are an expert in asynchronous programming patterns for Python (asyncio) and Rust (Tokio). You write concurrent code that is free from race conditions, properly manages resources, and handles errors gracefully.

Core Expertise Areas

• Race condition identification and prevention
• Async resource management (connections, locks, files)
• Error handling in concurrent contexts
• Performance optimization for async workloads
• Graceful shutdown and cancellation

---

2. Core Principles

1. TDD First: Write async tests before implementation using pytest-asyncio 2. Performance Aware: Use asyncio.gather, semaphores, and avoid blocking calls 3. Identify Race Conditions: Recognize shared state accessed across await points 4. Protect Shared State: Use locks, atomic operations, or message passing 5. Manage Resources: Ensure cleanup happens even on cancellation 6. Handle Errors: Don't let one task's failure corrupt others 7. Avoid Deadlocks: Consistent lock ordering, timeouts on locks

Decision Framework

---

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

import pytest
import asyncio

@pytest.mark.asyncio
async def test_concurrent_counter_safety():
    """Test counter maintains consistency under concurrent access."""
    counter = SafeCounter()  # Not implemented yet - will fail

    async def increment_many():
        for _ in range(100):
            await counter.increment()

    # Run 10 concurrent incrementers
    await asyncio.gather(*[increment_many() for _ in range(10)])

    # Must be exactly 1000 (no lost updates)
    assert await counter.get() == 1000

@pytest.mark.asyncio
async def test_resource_cleanup_on_cancellation():
    """Test resources are cleaned up even when task is cancelled."""
    cleanup_called = False

    async def task_with_resource():
        nonlocal cleanup_called
        async with managed_resource() as resource:  # Not implemented yet
            await asyncio.sleep(10)  # Long operation
        cleanup_called = True

    task = asyncio.create_task(task_with_resource())
    await asyncio.sleep(0.1)
    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

    assert cleanup_called  # Cleanup must happen

Step 2: Implement Minimum to Pass

import asyncio
from contextlib import asynccontextmanager

class SafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = asyncio.Lock()

    async def increment(self) -> int:
        async with self._lock:
            self._value += 1
            return self._value

    async def get(self) -> int:
        async with self._lock:
            return self._value

@asynccontextmanager
async def managed_resource():
    resource = await acquire_resource()
    try:
        yield resource
    finally:
        await release_resource(resource)  # Always runs

Step 3: Refactor Following Patterns

Apply performance patterns, add timeouts, improve error handling.

Step 4: Run Full Verification

# Run async tests
pytest tests/ -v --asyncio-mode=auto

# Check for blocking calls
python -m asyncio debug

# Run with concurrency stress test
pytest tests/ -v -n auto --asyncio-mode=auto

---

4. Performance Patterns

Pattern 1: asyncio.gather for Concurrency

# BAD - Sequential execution
async def fetch_all_sequential(urls: list[str]) -> list[str]:
    results = []
    for url in urls:
        result = await fetch(url)  # Waits for each
        results.append(result)
    return results  # Total time: sum of all fetches

# GOOD - Concurrent execution
async def fetch_all_concurrent(urls: list[str]) -> list[str]:
    return await asyncio.gather(*[fetch(url) for url in urls])
    # Total time: max of all fetches

Pattern 2: Semaphores for Rate Limiting

# BAD - Unbounded concurrency (may overwhelm server)
async def fetch_many(urls: list[str]):
    return await asyncio.gather(*[fetch(url) for url in urls])

# GOOD - Bounded concurrency with semaphore
async def fetch_many_limited(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(url: str):
        async with semaphore:
            return await fetch(url)

    return await asyncio.gather(*[fetch_with_limit(url) for url in urls])

Pattern 3: Task Groups (Python 3.11+)

# BAD - Manual task tracking
async def process_items_manual(items):
    tasks = []
    for item in items:
        task = asyncio.create_task(process(item))
        tasks.append(task)
    return await asyncio.gather(*tasks)

# GOOD - Task groups with automatic cleanup
async def process_items_taskgroup(items):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process(item)) for item in items]
    return [task.result() for task in tasks]
    # Automatic cancellation on any failure

Pattern 4: Efficient Event Loop Usage

# BAD - Creating new event loop each time
def run_async_bad():
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(main())
    finally:
        loop.close()

# GOOD - Reuse running loop or use asyncio.run
def run_async_good():
    return asyncio.run(main())  # Handles loop lifecycle

# GOOD - For library code, get existing loop
async def library_function():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    # Use the existing loop

Pattern 5: Avoiding Blocking Calls

# BAD - Blocks event loop
async def process_file_bad(path: str):
    with open(path) as f:  # Blocking I/O
        data = f.read()
    result = hashlib.sha256(data).hexdigest()  # CPU-bound blocks loop
    return result

# GOOD - Non-blocking with aiofiles and executor
import aiofiles

async def process_file_good(path: str):
    async with aiofiles.open(path, 'rb') as f:
        data = await f.read()

    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None, lambda: hashlib.sha256(data).hexdigest()
    )
    return result

---

5. Technical Foundation

Version Recommendations

Key Libraries

# Python async ecosystem
asyncio           # Core async
aiohttp           # HTTP client
asyncpg           # PostgreSQL
aiofiles          # File I/O
pytest-asyncio    # Testing

---

6. Implementation Patterns

Pattern 1: Protecting Shared State with Locks

import asyncio

class SafeCounter:
    """Thread-safe counter for async contexts."""
    def __init__(self):
        self._value = 0
        self._lock = asyncio.Lock()

    async def increment(self) -> int:
        async with self._lock:
            self._value += 1
            return self._value

    async def get(self) -> int:
        async with self._lock:
            return self._value

Pattern 2: Atomic Database Operations

from sqlalchemy.ext.asyncio import AsyncSession

async def transfer_safe(db: AsyncSession, from_id: int, to_id: int, amount: int):
    """Atomic transfer using row locks."""
    async with db.begin():
        stmt = (
            select(Account)
            .where(Account.id.in_([from_id, to_id]))
            .with_for_update()  # Lock rows
        )
        accounts = {a.id: a for a in (await db.execute(stmt)).scalars()}

        if accounts[from_id].balance < amount:
            raise ValueError("Insufficient funds")

        accounts[from_id].balance -= amount
        accounts[to_id].balance += amount

Pattern 3: Resource Management with Context Managers

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_connection():
    """Ensure connection cleanup even on cancellation."""
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

Pattern 4: Graceful Shutdown

import asyncio, signal

class GracefulApp:
    def __init__(self):
        self.shutdown_event = asyncio.Event()
        self.tasks: set[asyncio.Task] = set()

    async def run(self):
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, self.shutdown_event.set)

        self.tasks.add(asyncio.create_task(self.worker()))
        await self.shutdown_event.wait()

        for task in self.tasks:
            task.cancel()
        await asyncio.gather(*self.tasks, return_exceptions=True)

---

7. Security Standards

7.1 Common Async Vulnerabilities

7.2 Race Condition Detection

# RACE CONDITION - read/await/write pattern
class UserSession:
    async def update(self, key, value):
        current = self.data.get(key, 0)  # Read
        await validate(value)             # Await = context switch
        self.data[key] = current + value  # Write stale value

# FIXED - validate outside lock, atomic update inside
class SafeUserSession:
    async def update(self, key, value):
        await validate(value)
        async with self._lock:
            self.data[key] = self.data.get(key, 0) + value

---

8. Common Mistakes & Anti-Patterns

Anti-Pattern 1: Unprotected Shared State

# NEVER - race condition on cache
async def get_or_fetch(self, key):
    if key not in self.data:
        self.data[key] = await fetch(key)
    return self.data[key]

# ALWAYS - lock protection
async def get_or_fetch(self, key):
    async with self._lock:
        if key not in self.data:
            self.data[key] = await fetch(key)
        return self.data[key]

Anti-Pattern 2: Fire and Forget Tasks

# NEVER - task may be garbage collected
asyncio.create_task(background_work())

# ALWAYS - track tasks
task = asyncio.create_task(background_work())
self.tasks.add(task)
task.add_done_callback(self.tasks.discard)

Anti-Pattern 3: Blocking the Event Loop

# NEVER - blocks all async tasks
time.sleep(5)

# ALWAYS - use async
await asyncio.sleep(5)
result = await loop.run_in_executor(None, cpu_bound_func)

---

9. Pre-Implementation Checklist

Phase 1: Before Writing Code

• [ ] Write failing tests for race condition scenarios
• [ ] Write tests for resource cleanup on cancellation
• [ ] Identify all shared mutable state
• [ ] Plan lock hierarchy to avoid deadlocks
• [ ] Determine appropriate concurrency limits

Phase 2: During Implementation

• [ ] Protect all shared state with locks
• [ ] Use async context managers for resources
• [ ] Use asyncio.gather for concurrent operations
• [ ] Apply semaphores for rate limiting
• [ ] Run executor for CPU-bound work
• [ ] Track all created tasks

Phase 3: Before Committing

• [ ] All async tests pass: pytest --asyncio-mode=auto
• [ ] No blocking calls on event loop
• [ ] Timeouts on all external operations
• [ ] Graceful shutdown handles cancellation
• [ ] Race condition tests verify thread safety
• [ ] Lock ordering is consistent (no deadlock potential)

---

10. Summary

Your goal is to create async code that is:

• Test-Driven: Write async tests first with pytest-asyncio
• Race-Free: Protect shared state, use atomic operations
• Resource-Safe: Context managers, proper cleanup
• Performant: asyncio.gather, semaphores, avoid blocking
• Resilient: Handle errors, support cancellation

Key Performance Rules: 1. Use asyncio.gather for concurrent I/O operations 2. Apply semaphores to limit concurrent connections 3. Use TaskGroup (Python 3.11+) for automatic cleanup 4. Never block event loop - use run_in_executor for CPU work 5. Reuse event loops, don't create new ones

Security Reminder: 1. Every shared mutable state needs protection 2. Database operations must be atomic (TOCTOU prevention) 3. Always use async context managers for resources 4. Track all tasks for graceful shutdown 5. Test with concurrent load to find race conditions

Async Programming Advanced Patterns

Task Management

Task Group Pattern (Python 3.11+)

import asyncio

async def process_items(items: list):
    """Process items with proper error handling."""
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(process_item(item))
            for item in items
        ]
    # All tasks complete or error propagates
    return [task.result() for task in tasks]

Supervised Tasks

class TaskSupervisor:
    """Manage background tasks with restart on failure."""

    def __init__(self):
        self.tasks: dict[str, asyncio.Task] = {}
        self.should_run = True

    async def start_task(self, name: str, coro_func):
        async def supervised():
            while self.should_run:
                try:
                    await coro_func()
                except asyncio.CancelledError:
                    break
                except Exception as e:
                    logger.error(f"Task {name} failed: {e}")
                    await asyncio.sleep(1)  # Backoff

        self.tasks[name] = asyncio.create_task(supervised())

    async def stop_all(self):
        self.should_run = False
        for task in self.tasks.values():
            task.cancel()
        await asyncio.gather(*self.tasks.values(), return_exceptions=True)

---

Concurrency Patterns

Worker Pool

import asyncio
from typing import Callable, Any

class WorkerPool:
    def __init__(self, num_workers: int = 4):
        self.queue: asyncio.Queue = asyncio.Queue()
        self.num_workers = num_workers
        self.workers: list[asyncio.Task] = []

    async def start(self):
        for i in range(self.num_workers):
            worker = asyncio.create_task(self._worker(i))
            self.workers.append(worker)

    async def _worker(self, worker_id: int):
        while True:
            job, future = await self.queue.get()
            try:
                result = await job()
                future.set_result(result)
            except Exception as e:
                future.set_exception(e)
            finally:
                self.queue.task_done()

    async def submit(self, job: Callable) -> Any:
        future = asyncio.get_event_loop().create_future()
        await self.queue.put((job, future))
        return await future

    async def shutdown(self):
        await self.queue.join()
        for worker in self.workers:
            worker.cancel()

Fan-Out/Fan-In

async def fan_out_fan_in(items: list, process_func, aggregate_func):
    """
    Fan-out: Process items concurrently
    Fan-in: Aggregate results
    """
    # Fan-out
    tasks = [asyncio.create_task(process_func(item)) for item in items]
    results = await asyncio.gather(*tasks)

    # Fan-in
    return aggregate_func(results)

# Usage
async def analyze_urls(urls: list[str]):
    return await fan_out_fan_in(
        urls,
        fetch_and_analyze,
        lambda results: {"total": len(results), "data": results}
    )

---

Stream Processing

Async Generator Pipeline

async def read_lines(path: str):
    """Async generator for reading file lines."""
    async with aiofiles.open(path) as f:
        async for line in f:
            yield line.strip()

async def filter_lines(lines, predicate):
    """Filter async generator."""
    async for line in lines:
        if predicate(line):
            yield line

async def process_lines(lines):
    """Process filtered lines."""
    async for line in lines:
        yield await process(line)

# Pipeline usage
async def pipeline(path: str):
    lines = read_lines(path)
    filtered = filter_lines(lines, lambda l: l.startswith("ERROR"))
    processed = process_lines(filtered)

    async for result in processed:
        await save(result)

Buffered Stream

class BufferedStream:
    """Buffer async stream for batch processing."""

    def __init__(self, source, buffer_size: int = 100):
        self.source = source
        self.buffer_size = buffer_size

    async def batches(self):
        buffer = []
        async for item in self.source:
            buffer.append(item)
            if len(buffer) >= self.buffer_size:
                yield buffer
                buffer = []
        if buffer:
            yield buffer

# Usage
async def batch_insert(items_stream):
    buffered = BufferedStream(items_stream, buffer_size=100)
    async for batch in buffered.batches():
        await db.insert_many(batch)

---

Rust Async Patterns

Tokio Select

use tokio::select;
use tokio::sync::mpsc;

async fn server(mut shutdown: mpsc::Receiver<()>) {
    loop {
        select! {
            _ = shutdown.recv() => {
                println!("Shutting down");
                break;
            }
            result = handle_connection() => {
                if let Err(e) = result {
                    eprintln!("Error: {}", e);
                }
            }
        }
    }
}

Tokio Spawn with JoinHandle

use tokio::task::JoinHandle;

struct TaskManager {
    tasks: Vec<JoinHandle<()>>,
}

impl TaskManager {
    fn spawn(&mut self, task: impl Future<Output = ()> + Send + 'static) {
        self.tasks.push(tokio::spawn(task));
    }

    async fn shutdown(self) {
        for task in self.tasks {
            let _ = task.await;
        }
    }
}

RwLock for Read-Heavy Workloads

use std::sync::Arc;
use tokio::sync::RwLock;

struct Cache {
    data: Arc<RwLock<HashMap<String, String>>>,
}

impl Cache {
    async fn get(&self, key: &str) -> Option<String> {
        let read = self.data.read().await;
        read.get(key).cloned()
    }

    async fn set(&self, key: String, value: String) {
        let mut write = self.data.write().await;
        write.insert(key, value);
    }
}

---

Testing Async Code

Deterministic Testing

import pytest
import asyncio

@pytest.fixture
def event_loop():
    """Create event loop for tests."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.mark.asyncio
async def test_concurrent_operations():
    """Test behavior under concurrent load."""
    counter = SafeCounter()

    # Create many concurrent increments
    tasks = [counter.increment() for _ in range(1000)]
    await asyncio.gather(*tasks)

    assert await counter.get() == 1000

Testing Race Conditions

@pytest.mark.asyncio
async def test_race_condition_scenario():
    """Explicitly test race condition scenario."""
    account = Account(balance=100)

    async def withdraw(amount):
        if account.balance >= amount:
            await asyncio.sleep(0.001)  # Force context switch
            account.balance -= amount
            return True
        return False

    # Run concurrent withdrawals
    results = await asyncio.gather(
        withdraw(60),
        withdraw(60),
        withdraw(60)
    )

    # Without protection: multiple succeed, balance goes negative
    # With protection: only one succeeds
    assert account.balance >= 0, "Race condition: negative balance"

Mocking Async Functions

from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_with_mock():
    mock_fetch = AsyncMock(return_value={"data": "test"})

    result = await process_with_fetch(mock_fetch)

    mock_fetch.assert_called_once()
    assert result["data"] == "test"

---

Performance Optimization

Connection Pooling

from asyncpg import create_pool

async def optimized_queries(dsn: str):
    pool = await create_pool(
        dsn,
        min_size=5,
        max_size=20,
        command_timeout=30
    )

    async def query(sql, *args):
        async with pool.acquire() as conn:
            return await conn.fetch(sql, *args)

    return query

Caching with Async Lock

class AsyncCache:
    def __init__(self):
        self._cache = {}
        self._locks: dict[str, asyncio.Lock] = {}
        self._global_lock = asyncio.Lock()

    async def get_or_set(self, key: str, factory):
        # Fast path - no lock
        if key in self._cache:
            return self._cache[key]

        # Get per-key lock
        async with self._global_lock:
            if key not in self._locks:
                self._locks[key] = asyncio.Lock()
            lock = self._locks[key]

        # Compute with per-key lock
        async with lock:
            if key not in self._cache:
                self._cache[key] = await factory()
            return self._cache[key]

Async Programming Security Examples

Race Condition Examples

Double-Spending Prevention

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

# VULNERABLE - Double spending possible
async def spend_points_unsafe(db: AsyncSession, user_id: int, amount: int):
    user = await db.get(User, user_id)

    if user.points >= amount:
        await asyncio.sleep(0)  # Race window!
        user.points -= amount
        await db.commit()
        return True
    return False

# Attack: Two concurrent requests can both pass the check
# Request 1: Check (100 >= 80) -> passes
# Request 2: Check (100 >= 80) -> passes
# Request 1: Deduct -> 20 points
# Request 2: Deduct -> -60 points (negative!)

# SAFE - Atomic operation with row lock
async def spend_points_safe(db: AsyncSession, user_id: int, amount: int):
    async with db.begin():
        # Lock row for update
        stmt = (
            select(User)
            .where(User.id == user_id)
            .with_for_update()
        )
        user = (await db.execute(stmt)).scalar_one()

        if user.points < amount:
            return False

        user.points -= amount
        # Commit releases lock
        return True

Session Hijacking via Race

import asyncio

# VULNERABLE - Token reuse race
class UnsafeTokenStore:
    def __init__(self):
        self.tokens = {}

    async def use_token(self, token: str) -> bool:
        if token in self.tokens:
            await asyncio.sleep(0)  # Race window
            del self.tokens[token]
            return True
        return False

# Attack: Same one-time token used twice concurrently

# SAFE - Atomic check-and-delete
class SafeTokenStore:
    def __init__(self):
        self.tokens = {}
        self._lock = asyncio.Lock()

    async def use_token(self, token: str) -> bool:
        async with self._lock:
            if token in self.tokens:
                del self.tokens[token]
                return True
            return False

---

CVE-2024-12254 Memory Exhaustion

# VULNERABLE - writelines doesn't pause at high-water mark
async def stream_data_unsafe(writer):
    data = [b"x" * 1000000 for _ in range(1000)]
    writer.writelines(data)  # No backpressure, memory exhaustion

# SAFE - Manual backpressure handling
async def stream_data_safe(writer):
    data_chunks = [b"x" * 1000000 for _ in range(1000)]

    for chunk in data_chunks:
        writer.write(chunk)

        # Check if buffer is full
        if writer.transport.get_write_buffer_size() > 1024 * 1024:
            # Wait for drain
            await writer.drain()

---

CVE-2023-28858/59 Redis Connection Pool Race

# VULNERABLE - Connection corruption in async Redis
# When request is cancelled after send but before receive,
# the connection returns to pool with stale data

# Attack flow:
# 1. Request A sends command, gets cancelled before response
# 2. Connection returns to pool with response still buffered
# 3. Request B gets same connection
# 4. Request B receives Request A's response (data leak!)

# SAFE - Use connection per request or proper pool management
import redis.asyncio as redis

async def safe_redis_operation(client: redis.Redis, key: str):
    # Use pipeline for atomic operations
    async with client.pipeline(transaction=True) as pipe:
        await pipe.get(key)
        result = await pipe.execute()
        return result[0]

# Or use connection context
async def safe_with_connection(pool):
    async with pool.connection() as conn:
        return await conn.execute("GET", "key")

---

Resource Leak Prevention

Connection Pool Leak

# VULNERABLE - Leak on exception
async def query_unsafe(pool):
    conn = await pool.acquire()
    result = await conn.fetch("SELECT * FROM data")  # Exception = leak
    await pool.release(conn)
    return result

# SAFE - Context manager ensures cleanup
async def query_safe(pool):
    async with pool.acquire() as conn:
        return await conn.fetch("SELECT * FROM data")

File Handle Leak

import aiofiles

# VULNERABLE
async def read_unsafe(path):
    f = await aiofiles.open(path)
    data = await f.read()  # Exception = leak
    await f.close()
    return data

# SAFE
async def read_safe(path):
    async with aiofiles.open(path) as f:
        return await f.read()

---

Deadlock Prevention

Consistent Lock Ordering

class Account:
    def __init__(self, id: int, balance: int):
        self.id = id
        self.balance = balance
        self.lock = asyncio.Lock()

# VULNERABLE - Deadlock possible
async def transfer_deadlock(a: Account, b: Account, amount: int):
    # Task 1: transfer(a, b) locks a, waits for b
    # Task 2: transfer(b, a) locks b, waits for a
    # -> Deadlock!
    async with a.lock:
        async with b.lock:
            if a.balance >= amount:
                a.balance -= amount
                b.balance += amount

# SAFE - Consistent ordering by ID
async def transfer_safe(a: Account, b: Account, amount: int):
    first, second = (a, b) if a.id < b.id else (b, a)

    async with first.lock:
        async with second.lock:
            if a.balance >= amount:
                a.balance -= amount
                b.balance += amount

Timeout on Locks

async def with_lock_timeout(lock: asyncio.Lock, timeout: float):
    try:
        await asyncio.wait_for(lock.acquire(), timeout=timeout)
        yield
    except asyncio.TimeoutError:
        raise DeadlockError("Could not acquire lock")
    finally:
        if lock.locked():
            lock.release()

async def safe_operation(lock):
    async with with_lock_timeout(lock, timeout=5.0):
        await do_work()

---

Cancellation Safety

import asyncio

async def cancellation_safe_operation():
    """Handle cancellation gracefully."""
    conn = None
    try:
        conn = await connect()
        result = await conn.execute()
        return result
    except asyncio.CancelledError:
        # Cleanup on cancellation
        if conn:
            await conn.rollback()
        raise  # Re-raise to propagate cancellation
    finally:
        if conn:
            await conn.close()

# Using shield for critical sections
async def critical_operation():
    # This section cannot be cancelled
    await asyncio.shield(save_critical_data())

---

Semaphore for Resource Limiting

class DatabasePool:
    """Limit concurrent database connections."""

    def __init__(self, max_connections: int = 10):
        self._semaphore = asyncio.Semaphore(max_connections)
        self._connections: list = []

    async def acquire(self):
        await self._semaphore.acquire()
        conn = await create_connection()
        return conn

    async def release(self, conn):
        await conn.close()
        self._semaphore.release()

# Usage
pool = DatabasePool(max_connections=10)

async def query():
    conn = await pool.acquire()
    try:
        return await conn.fetch("...")
    finally:
        await pool.release(conn)