Skip to main content

Concurrency

In this chapter, we'll explore concurrent programming in Python. We'll cover:

  • Threading basics
  • The Global Interpreter Lock (GIL)
  • asyncio for asynchronous programming
  • async and await keywords
  • Testing concurrent code
  • When to use threading vs asyncio

Why concurrency?​

Concurrency allows programs to handle multiple tasks at once:

  • I/O-bound tasks: Waiting for network, disk, or user input
  • CPU-bound tasks: Heavy computation (better with multiprocessing)

Python offers two main approaches: threading and asyncio.

Threading basics​

A simple threaded example​

import threading
import time


def download_file(url):
    print(f"Starting download: {url}")
    time.sleep(2)  # Simulate download
    print(f"Finished download: {url}")


# Sequential (slow)
for url in ["file1.txt", "file2.txt", "file3.txt"]:
    download_file(url)
# Takes ~6 seconds

# Concurrent with threads (fast)
threads = []
for url in ["file1.txt", "file2.txt", "file3.txt"]:
    t = threading.Thread(target=download_file, args=(url,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()  # Wait for all threads to complete
# Takes ~2 seconds

Testing threaded code​

import threading
from concurrent import ConcurrentCounter


def test_concurrent_counter():
    counter = ConcurrentCounter()
    
    def increment_many():
        for _ in range(1000):
            counter.increment()
    
    threads = [threading.Thread(target=increment_many) for _ in range(10)]
    
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    
    assert counter.value == 10000

Thread-safe implementation with Lock​

import threading


class ConcurrentCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()
    
    def increment(self):
        with self._lock:
            self._value += 1
    
    @property
    def value(self):
        with self._lock:
            return self._value

The Lock ensures only one thread modifies _value at a time.

The Global Interpreter Lock (GIL)​

Python's GIL allows only one thread to execute Python bytecode at a time. This means:

  • I/O-bound tasks: Threading works well (threads release GIL during I/O)
  • CPU-bound tasks: Threading doesn't help (use multiprocessing instead)
import threading
import time


def cpu_intensive():
    total = 0
    for i in range(10_000_000):
        total += i
    return total


# Threading won't speed this up due to GIL
start = time.time()
threads = [threading.Thread(target=cpu_intensive) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Threaded: {time.time() - start:.2f}s")

For CPU-bound work, use multiprocessing or concurrent.futures.ProcessPoolExecutor.

ThreadPoolExecutor​

A higher-level interface for threading:

from concurrent.futures import ThreadPoolExecutor
import time


def fetch_url(url):
    time.sleep(1)  # Simulate network request
    return f"Content from {url}"


urls = ["url1", "url2", "url3", "url4"]

# Using ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch_url, urls))

print(results)  # Takes ~1 second instead of ~4

Testing with ThreadPoolExecutor​

from concurrent.futures import ThreadPoolExecutor
from url_fetcher import fetch_all


def test_fetch_all():
    urls = ["http://example1.com", "http://example2.com"]
    
    results = fetch_all(urls)
    
    assert len(results) == 2
    assert all("Content" in r for r in results)

asyncio basics​

asyncio is Python's native async library. It uses a single thread with an event loop.

Coroutines with async/await​

import asyncio


async def say_hello(name, delay):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")


async def main():
    # Run coroutines concurrently
    await asyncio.gather(
        say_hello("Alice", 1),
        say_hello("Bob", 2),
        say_hello("Charlie", 1.5),
    )

asyncio.run(main())
# All complete in ~2 seconds

Key concepts​

  • async def - Defines a coroutine
  • await - Pauses execution until the awaited coroutine completes
  • asyncio.run() - Runs the event loop
  • asyncio.gather() - Runs multiple coroutines concurrently

Testing async code​

pytest has built-in support for async tests with pytest-asyncio:

pip install pytest-asyncio

Writing async tests​

import pytest
from async_service import AsyncUserService


@pytest.mark.asyncio
async def test_get_user():
    service = AsyncUserService()
    
    user = await service.get_user(123)
    
    assert user["id"] == 123
    assert user["name"] == "Alice"

Testing concurrent operations​

import pytest
import asyncio
from async_downloader import download_all


@pytest.mark.asyncio
async def test_download_all():
    urls = ["url1", "url2", "url3"]
    
    results = await download_all(urls)
    
    assert len(results) == 3
    assert all(r.startswith("Content") for r in results)

Building an async service​

Write the test first​

import pytest
from async_weather import WeatherService


class FakeWeatherAPI:
    async def get_temperature(self, city):
        await asyncio.sleep(0.1)  # Simulate network
        temperatures = {
            "London": 15.0,
            "Paris": 18.0,
            "Tokyo": 22.0,
        }
        return temperatures.get(city, 0.0)


@pytest.mark.asyncio
async def test_get_temperatures():
    fake_api = FakeWeatherAPI()
    service = WeatherService(fake_api)
    
    cities = ["London", "Paris", "Tokyo"]
    temps = await service.get_temperatures(cities)
    
    assert temps == {"London": 15.0, "Paris": 18.0, "Tokyo": 22.0}

Implementation​

import asyncio


class WeatherService:
    def __init__(self, api):
        self.api = api
    
    async def get_temperatures(self, cities):
        # Fetch all temperatures concurrently
        tasks = [self.api.get_temperature(city) for city in cities]
        temps = await asyncio.gather(*tasks)
        return dict(zip(cities, temps))

Error handling in async code​

async def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await fetch_url(url)
        except ConnectionError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

Testing error handling​

import pytest


@pytest.mark.asyncio
async def test_fetch_with_retry_fails():
    async def failing_fetch(url):
        raise ConnectionError("Network error")
    
    with pytest.raises(ConnectionError):
        await fetch_with_retry("http://example.com")

Async context managers​

class AsyncDatabase:
    async def __aenter__(self):
        await self.connect()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()
    
    async def connect(self):
        print("Connecting...")
    
    async def disconnect(self):
        print("Disconnecting...")


# Usage
async def main():
    async with AsyncDatabase() as db:
        await db.query("SELECT * FROM users")

When to use what​

Use threading when​

  • You have I/O-bound tasks (network, file I/O)
  • You're working with existing synchronous code
  • Libraries don't support async
  • You need simple parallelism

Use asyncio when​

  • Building new I/O-bound applications
  • You need many concurrent connections
  • Using async-compatible libraries
  • Building web servers or API clients

Use multiprocessing when​

  • You have CPU-bound tasks
  • You need true parallelism (bypass GIL)
  • Tasks are independent and don't share state

A complete example: Async web scraper​

# test_scraper.py
import pytest
from scraper import WebScraper


class FakeHttpClient:
    def __init__(self, responses):
        self.responses = responses
    
    async def get(self, url):
        return self.responses.get(url, "Not found")


@pytest.mark.asyncio
async def test_scrape_multiple_pages():
    responses = {
        "http://site1.com": "<html>Site 1</html>",
        "http://site2.com": "<html>Site 2</html>",
    }
    fake_client = FakeHttpClient(responses)
    scraper = WebScraper(fake_client)
    
    results = await scraper.scrape([
        "http://site1.com",
        "http://site2.com",
    ])
    
    assert len(results) == 2
    assert "Site 1" in results["http://site1.com"]
    assert "Site 2" in results["http://site2.com"]

Implementation:

# scraper.py
import asyncio


class WebScraper:
    def __init__(self, http_client):
        self.client = http_client
    
    async def scrape(self, urls):
        tasks = [self._scrape_one(url) for url in urls]
        results = await asyncio.gather(*tasks)
        return dict(zip(urls, results))
    
    async def _scrape_one(self, url):
        return await self.client.get(url)

Wrapping up​

We've covered:

  • Threading - Run code in separate threads with threading.Thread
  • Locks - Protect shared state with threading.Lock
  • GIL - Understanding Python's Global Interpreter Lock
  • ThreadPoolExecutor - Higher-level thread management
  • asyncio - Event loop-based concurrency
  • async/await - Syntax for asynchronous code
  • Testing - Use pytest-asyncio for async tests

Key takeaways​

  1. Use threading for I/O-bound tasks with existing sync code
  2. Use asyncio for new I/O-bound applications
  3. Use multiprocessing for CPU-bound tasks
  4. Always use locks when sharing state between threads
  5. Test concurrent code carefully - race conditions are tricky!

Concurrency adds complexity, but when used correctly, it can dramatically improve performance for I/O-bound applications.