Async/Await & Concurrency: Parallel Python

import time

def fetch_data(url: str) -> str:
    """Simulate a slow network request."""
    print(f"Fetching {url}...")
    time.sleep(2)  # simulate 2-second network delay
    return f"Data from {url}"

# Sequential: each request waits for the previous one to finish
start = time.time()

result1 = fetch_data("api.example.com/users")
result2 = fetch_data("api.example.com/orders")
result3 = fetch_data("api.example.com/products")

elapsed = time.time() - start
print(f"\nTotal time: {elapsed:.1f}s")  # ~6 seconds (2+2+2)

import threading
import time

def download_file(filename: str, seconds: float) -> None:
    """Simulate downloading a file."""
    thread_name = threading.current_thread().name
    print(f"[{thread_name}] Downloading {filename}...")
    time.sleep(seconds)  # simulate I/O wait
    print(f"[{thread_name}] Finished {filename}")

# Create and start threads
start = time.time()

thread1 = threading.Thread(target=download_file, args=("data.csv", 2), name="T1")
thread2 = threading.Thread(target=download_file, args=("model.pkl", 3), name="T2")
thread3 = threading.Thread(target=download_file, args=("config.json", 1), name="T3")

thread1.start()
thread2.start()
thread3.start()

# Wait for all threads to finish
thread1.join()
thread2.join()
thread3.join()

elapsed = time.time() - start
print(f"\nTotal time: {elapsed:.1f}s")  # ~3 seconds (not 6!)

import threading
import time

def count_up(n: int) -> int:
    """CPU-bound: count to n."""
    total = 0
    for i in range(n):
        total += i
    return total

N = 50_000_000

# Sequential
start = time.time()
count_up(N)
count_up(N)
sequential_time = time.time() - start

# Threaded (same work, but in threads)
start = time.time()
t1 = threading.Thread(target=count_up, args=(N,))
t2 = threading.Thread(target=count_up, args=(N,))
t1.start()
t2.start()
t1.join()
t2.join()
threaded_time = time.time() - start

print(f"Sequential: {sequential_time:.2f}s")
print(f"Threaded:   {threaded_time:.2f}s")
# Both take roughly the same time! The GIL prevents true parallel execution.

import threading

# UNSAFE: race condition
class UnsafeCounter:
    def __init__(self):
        self.count = 0

    def increment(self):
        # This is NOT atomic! It's: read count, add 1, write count
        # Another thread can read between the add and write
        self.count += 1

# SAFE: using a lock
class SafeCounter:
    def __init__(self):
        self.count = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:  # only one thread can enter this block at a time
            self.count += 1

def run_increments(counter, n: int) -> None:
    for _ in range(n):
        counter.increment()

# Test both counters
for CounterClass in [UnsafeCounter, SafeCounter]:
    counter = CounterClass()
    threads = [
        threading.Thread(target=run_increments, args=(counter, 100_000))
        for _ in range(10)
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    expected = 1_000_000
    actual = counter.count
    name = CounterClass.__name__
    print(f"{name}: expected={expected}, actual={actual}, correct={expected == actual}")

# UnsafeCounter: expected=1000000, actual=987432, correct=False  (varies!)
# SafeCounter:   expected=1000000, actual=1000000, correct=True

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url: str) -> dict:
    """Simulate fetching data from a URL."""
    time.sleep(1)  # simulate network latency
    return {"url": url, "status": 200, "data": f"Response from {url}"}

urls = [
    "api.example.com/users",
    "api.example.com/orders",
    "api.example.com/products",
    "api.example.com/analytics",
    "api.example.com/settings",
]

# ThreadPoolExecutor manages a pool of worker threads
start = time.time()

with ThreadPoolExecutor(max_workers=5) as executor:
    # map() applies the function to each item and returns results in order
    results = list(executor.map(fetch_url, urls))

elapsed = time.time() - start
print(f"Fetched {len(results)} URLs in {elapsed:.1f}s")  # ~1 second, not 5
for r in results:
    print(f"  {r['url']}: {r['status']}")

import multiprocessing
import time
import math

def compute_heavy(n: int) -> float:
    """CPU-intensive computation: sum of square roots."""
    return sum(math.sqrt(i) for i in range(n))

N = 10_000_000

# Sequential
start = time.time()
r1 = compute_heavy(N)
r2 = compute_heavy(N)
r3 = compute_heavy(N)
r4 = compute_heavy(N)
sequential_time = time.time() - start

# Parallel with multiprocessing
start = time.time()
with multiprocessing.Pool(processes=4) as pool:
    results = pool.map(compute_heavy, [N, N, N, N])
parallel_time = time.time() - start

print(f"Sequential: {sequential_time:.2f}s")
print(f"Parallel:   {parallel_time:.2f}s")
print(f"Speedup:    {sequential_time / parallel_time:.1f}x")
# On a 4-core machine: roughly 3.5-4x speedup

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n: int) -> bool:
    """Check if n is prime (CPU-bound)."""
    if n < 2:
        return False
    if n < 4:
        return True
    if n % 2 == 0 or n % 3 == 0:
        return False
    i = 5
    while i * i <= n:
        if n % i == 0 or n % (i + 2) == 0:
            return False
        i += 6
    return True

# Check many large numbers for primality in parallel
numbers = [
    112272535095293, 112582705942171, 112272535095293,
    115280095190773, 115797848077099, 1099726899285419,
    # ... imagine thousands of these
]

# Sequential
import time
start = time.time()
seq_results = [is_prime(n) for n in numbers]
seq_time = time.time() - start

# Parallel
start = time.time()
with ProcessPoolExecutor() as executor:
    par_results = list(executor.map(is_prime, numbers))
par_time = time.time() - start

print(f"Sequential: {seq_time:.4f}s")
print(f"Parallel:   {par_time:.4f}s")
for n, prime in zip(numbers, par_results):
    print(f"  {n}: {'prime' if prime else 'not prime'}")

import multiprocessing

def worker(shared_counter, lock, n: int) -> None:
    """Increment a shared counter n times."""
    for _ in range(n):
        with lock:
            shared_counter.value += 1

if __name__ == "__main__":
    counter = multiprocessing.Value("i", 0)  # shared integer, initial value 0
    lock = multiprocessing.Lock()

    processes = [
        multiprocessing.Process(target=worker, args=(counter, lock, 100_000))
        for _ in range(4)
    ]

    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print(f"Final count: {counter.value}")  # 400000

import asyncio

# An async function (coroutine) -- defined with 'async def'
async def fetch_data(name: str, delay: float) -> str:
    """Simulate an async network request."""
    print(f"  Starting {name}...")
    await asyncio.sleep(delay)  # non-blocking sleep (yields control to event loop)
    print(f"  Finished {name}")
    return f"Data from {name}"

async def main():
    """Run multiple async tasks concurrently."""
    # Use the running loop directly — `asyncio.get_event_loop()` is deprecated
    # in Python 3.12+ when called without a running loop.
    loop = asyncio.get_running_loop()
    start = loop.time()

    # asyncio.gather() runs coroutines concurrently
    results = await asyncio.gather(
        fetch_data("users", 2),
        fetch_data("orders", 3),
        fetch_data("products", 1),
    )

    elapsed = loop.time() - start
    print(f"\nAll done in {elapsed:.1f}s")  # ~3 seconds (not 6!)
    for r in results:
        print(f"  {r}")

# Run the async main function
asyncio.run(main())

import asyncio

async def background_job(name: str, seconds: float) -> str:
    """A task that runs in the background."""
    print(f"  [{name}] started")
    await asyncio.sleep(seconds)
    print(f"  [{name}] completed")
    return f"Result from {name}"

async def main():
    # Create tasks — they start running immediately
    task1 = asyncio.create_task(background_job("download", 3))
    task2 = asyncio.create_task(background_job("process", 2))
    task3 = asyncio.create_task(background_job("upload", 1))

    # Do other work while tasks run in background
    print("Main: doing other work while tasks run...")
    await asyncio.sleep(0.5)
    print("Main: still working...")

    # Now wait for all tasks to complete
    results = await asyncio.gather(task1, task2, task3)
    print(f"\nAll results: {results}")

asyncio.run(main())

import asyncio

class AsyncTimer:
    """An async context manager for timing async operations."""

    async def __aenter__(self):
        # Inside a coroutine, the loop is always running — use get_running_loop().
        self._loop = asyncio.get_running_loop()
        self.start = self._loop.time()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        elapsed = self._loop.time() - self.start
        print(f"Async operation took {elapsed:.2f}s")

async def fetch_batch(items: list[str]) -> list[str]:
    """Fetch multiple items with timing."""
    async with AsyncTimer():
        tasks = [asyncio.create_task(fetch_item(item)) for item in items]
        return await asyncio.gather(*tasks)

async def fetch_item(item: str) -> str:
    await asyncio.sleep(0.5)
    return f"fetched:{item}"

async def main():
    results = await fetch_batch(["a", "b", "c", "d", "e"])
    print(results)

asyncio.run(main())

import asyncio

async def async_range(start: int, stop: int, delay: float):
    """An async generator that yields numbers with a delay."""
    for i in range(start, stop):
        await asyncio.sleep(delay)
        yield i

async def main():
    # Async for loop
    async for num in async_range(0, 5, 0.3):
        print(f"Got: {num}")

asyncio.run(main())

import asyncio

async def risky_fetch(url: str) -> str:
    """A fetch that might fail."""
    await asyncio.sleep(0.5)
    if "bad" in url:
        raise ConnectionError(f"Failed to connect to {url}")
    return f"Data from {url}"

async def main():
    # gather with return_exceptions=True collects errors instead of raising
    results = await asyncio.gather(
        risky_fetch("api.example.com/users"),
        risky_fetch("bad.example.com/data"),
        risky_fetch("api.example.com/orders"),
        return_exceptions=True,
    )

    for r in results:
        if isinstance(r, Exception):
            print(f"  ERROR: {r}")
        else:
            print(f"  OK: {r}")

asyncio.run(main())
# OK: Data from api.example.com/users
# ERROR: Failed to connect to bad.example.com/data
# OK: Data from api.example.com/orders

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
import math

# I/O-bound task
def fetch_page(url: str) -> str:
    time.sleep(1)  # simulate network I/O
    return f"Page content from {url}"

# CPU-bound task
def compute_primes(limit: int) -> int:
    count = 0
    for n in range(2, limit):
        if all(n % i != 0 for i in range(2, int(math.sqrt(n)) + 1)):
            count += 1
    return count

urls = [f"example.com/page/{i}" for i in range(10)]
limits = [100_000] * 4

# I/O-bound: use ThreadPoolExecutor
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    page_results = list(executor.map(fetch_page, urls))
print(f"Thread pool (I/O): {time.time() - start:.1f}s for {len(urls)} pages")

# CPU-bound: use ProcessPoolExecutor
start = time.time()
with ProcessPoolExecutor(max_workers=4) as executor:
    prime_counts = list(executor.map(compute_primes, limits))
print(f"Process pool (CPU): {time.time() - start:.1f}s for {len(limits)} computations")

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def process_item(item_id: int) -> dict:
    """Process an item with variable latency."""
    delay = random.uniform(0.5, 3.0)
    time.sleep(delay)
    return {"id": item_id, "delay": round(delay, 2), "status": "done"}

with ThreadPoolExecutor(max_workers=5) as executor:
    # submit() returns a Future object immediately
    futures = {
        executor.submit(process_item, i): i
        for i in range(10)
    }

    # as_completed() yields futures as they finish (fastest first)
    for future in as_completed(futures):
        item_id = futures[future]
        try:
            result = future.result()
            print(f"Item {result['id']} completed in {result['delay']}s")
        except Exception as e:
            print(f"Item {item_id} failed: {e}")

from concurrent.futures import ThreadPoolExecutor
import time

def slow_add(a: int, b: int) -> int:
    time.sleep(1)
    return a + b

with ThreadPoolExecutor() as executor:
    future = executor.submit(slow_add, 3, 7)

    print(f"Done? {future.done()}")      # False (still running)
    print(f"Running? {future.running()}") # True

    result = future.result(timeout=5)     # blocks until done (with timeout)
    print(f"Result: {result}")            # 10
    print(f"Done? {future.done()}")       # True

    # You can also add callbacks
    future2 = executor.submit(slow_add, 10, 20)
    future2.add_done_callback(
        lambda f: print(f"Callback: result is {f.result()}")
    )

# Decision tree for choosing concurrency approach:
#
# Is your task I/O-bound or CPU-bound?
# ├── I/O-bound (network, disk, database)
# │   ├── Need thousands of concurrent connections? → asyncio
# │   ├── Simple parallel fetching? → ThreadPoolExecutor
# │   └── Working with sync-only libraries? → ThreadPoolExecutor
# │
# └── CPU-bound (math, image processing, ML training)
#     ├── Can use NumPy/PyTorch (releases GIL)? → ThreadPoolExecutor works
#     ├── Pure Python computation? → ProcessPoolExecutor
#     └── Need shared state? → multiprocessing with Manager/Value/Queue

import asyncio

# Simulated aiohttp-style client
class AsyncAPIClient:
    """A simple async API client pattern."""

    def __init__(self, base_url: str, max_concurrent: int = 10):
        self.base_url = base_url
        self.semaphore = asyncio.Semaphore(max_concurrent)  # limit concurrency

    async def fetch(self, endpoint: str) -> dict:
        """Fetch a single endpoint with rate limiting."""
        async with self.semaphore:  # at most max_concurrent requests at once
            # In real code: async with aiohttp.ClientSession() as session:
            #     async with session.get(f"{self.base_url}/{endpoint}") as resp:
            #         return await resp.json()
            await asyncio.sleep(0.5)  # simulate network latency
            return {"endpoint": endpoint, "status": 200}

    async def fetch_all(self, endpoints: list[str]) -> list[dict]:
        """Fetch multiple endpoints concurrently."""
        tasks = [self.fetch(ep) for ep in endpoints]
        return await asyncio.gather(*tasks)

async def main():
    client = AsyncAPIClient("https://api.example.com", max_concurrent=5)

    endpoints = [f"users/{i}" for i in range(20)]
    results = await client.fetch_all(endpoints)
    print(f"Fetched {len(results)} endpoints")
    print(f"Sample: {results[0]}")

asyncio.run(main())

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time
import json

def download_chunk(chunk_url: str) -> list[dict]:
    """I/O-bound: download a data chunk."""
    time.sleep(0.5)  # simulate download
    return [{"id": i, "value": i * 2.5} for i in range(1000)]

def process_chunk(records: list[dict]) -> dict:
    """CPU-bound: aggregate records."""
    total = sum(r["value"] for r in records)
    count = len(records)
    return {"count": count, "total": total, "mean": total / count}

def parallel_pipeline(chunk_urls: list[str]) -> list[dict]:
    """Two-stage pipeline: download (I/O) then process (CPU)."""
    # Stage 1: Download in parallel using threads (I/O-bound)
    with ThreadPoolExecutor(max_workers=8) as io_pool:
        raw_chunks = list(io_pool.map(download_chunk, chunk_urls))

    # Stage 2: Process in parallel using processes (CPU-bound)
    with ProcessPoolExecutor(max_workers=4) as cpu_pool:
        results = list(cpu_pool.map(process_chunk, raw_chunks))

    return results

if __name__ == "__main__":
    urls = [f"data.example.com/chunk/{i}" for i in range(20)]

    start = time.time()
    results = parallel_pipeline(urls)
    elapsed = time.time() - start

    total_records = sum(r["count"] for r in results)
    print(f"Processed {total_records:,} records in {elapsed:.1f}s")
    print(f"Chunks processed: {len(results)}")

import asyncio
from concurrent.futures import ProcessPoolExecutor

# Simulated ML model
class SimpleModel:
    """A simulated ML model for demonstration."""

    def predict(self, inputs: list[float]) -> list[float]:
        """CPU-bound prediction."""
        import math
        return [math.tanh(x) for x in inputs]

def run_inference(batch: list[float]) -> list[float]:
    """Run model inference on a batch (CPU-bound, runs in separate process)."""
    model = SimpleModel()
    return model.predict(batch)

async def async_inference_server(request_queue: list[list[float]]) -> list[list[float]]:
    """
    Async server that:
    1. Receives requests asynchronously (I/O-bound)
    2. Batches them
    3. Runs inference in a process pool (CPU-bound)
    """
    loop = asyncio.get_running_loop()  # use get_running_loop() inside coroutines

    with ProcessPoolExecutor(max_workers=4) as pool:
        # Run CPU-bound inference in process pool from async context
        futures = [
            loop.run_in_executor(pool, run_inference, batch)
            for batch in request_queue
        ]
        results = await asyncio.gather(*futures)

    return results

async def main():
    # Simulate incoming prediction requests
    batches = [
        [0.1, 0.5, 0.9, -0.3, 1.2],
        [2.0, -1.5, 0.0, 0.7, -0.8],
        [1.1, -2.2, 3.3, -4.4, 5.5],
        [0.0, 0.0, 0.0, 0.0, 0.0],
    ]

    results = await async_inference_server(batches)
    for i, (batch, preds) in enumerate(zip(batches, results)):
        print(f"Batch {i}: {batch[:3]}... -> {[f'{p:.3f}' for p in preds[:3]]}...")

asyncio.run(main())

import asyncio

async def scrape_page(url: str, semaphore: asyncio.Semaphore) -> dict:
    """Scrape a single page with rate limiting."""
    async with semaphore:
        print(f"  Scraping {url}...")
        await asyncio.sleep(0.3)  # simulate network request
        return {
            "url": url,
            "title": f"Page at {url}",
            "word_count": 500,
        }

async def scrape_site(base_url: str, num_pages: int, max_concurrent: int = 5) -> list:
    """Scrape an entire site with controlled concurrency."""
    semaphore = asyncio.Semaphore(max_concurrent)
    urls = [f"{base_url}/page/{i}" for i in range(num_pages)]

    tasks = [scrape_page(url, semaphore) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]

    print(f"\nScraped {len(successes)} pages, {len(failures)} failures")
    return successes

async def main():
    results = await scrape_site("https://example.com", num_pages=20, max_concurrent=5)
    total_words = sum(r["word_count"] for r in results)
    print(f"Total words scraped: {total_words:,}")

asyncio.run(main())