Concurrency, Parallelism & Performance

A job is too slow. Your first idea may be to add threads. That is the wrong first step.

Speed work has an order. You will use it in this chapter:

1. Measure, to find the slow part.
2. Fix the algorithm and cache repeated work.
3. Pick the cheapest concurrency model that fits the slow part.
4. Move to many machines (a cluster) only if one machine cannot do the job.

Two words come first. Concurrency means many tasks are in progress, and one worker switches between them. Parallelism means many tasks run at the same instant.

Parallelism needs many CPU cores. A core is one part of the CPU that runs code. Concurrency works on one core.

Next, ask where the time goes. If the task waits, it is I/O-bound. I/O means input and output, such as network and disk.

If the task computes, it is CPU-bound.

A checksum is a short code made from file data. Two files with the same data give the same code.

The shop gives us one task of each kind. An API (application programming interface) is a service that your code calls over the network.

Look at the two tasks in this table:

The kind of task decides the tool. A concurrency model is a way to run many tasks: threads, processes or asyncio.

Profiling means measuring which part of a program uses the time. Without it, you guess.

Python has two tools. time.perf_counter() is a stopwatch for one block of code. cProfile counts and times every function call.

This example profiles a small total. The numbers of calls are the same on every run:

import cProfile
import pstats

def parse(line):
    return float(line.split(",")[3])

def total(lines):
    return sum(parse(line) for line in lines)

lines = ["101,1,paid,40.0"] * 1000

profiler = cProfile.Profile()
profiler.enable()
total(lines)
profiler.disable()

calls = pstats.Stats(profiler).get_stats_profile().func_profiles
print("total", calls["total"].ncalls)
print("parse", calls["parse"].ncalls)

cProfile also records the time for each function. Time changes on every run, so we do not print it here.

To see it, call pstats.Stats(profiler).sort_stats("tottime").print_stats(5). tottime is the time spent inside a function itself, not in the functions it calls. The function at the top has the most tottime.

Fix that function first. If the top function waits on the network, think about threads. If it computes, read the next section.

Concurrency is rarely the first fix. Three fixes are cheaper, and they do not make workers share data.

First, the algorithm. A list scan checks every item. A set finds an item in about one step, however big the set is.

Checking an order_id in a set of 1 million ids is about as fast as in a set of 10.

Second, cache repeated work. functools.lru_cache remembers the result of a call. This code looks up a customer tier for each order:

from functools import lru_cache

lookups = []

@lru_cache(maxsize=None)
def customer_tier(customer_id):
    lookups.append(customer_id)   # stands in for a slow lookup
    return "gold" if customer_id in (1, 4) else "basic"

for customer_id in [1, 1, 2, 3, 3, 4]:
    customer_tier(customer_id)

print(lookups)
print(customer_tier.cache_info())

Third, stream big files and move loops out of Python. A generator reads a big file one row at a time, as in Iterators, Generators & Streaming Data. A numpy or pandas operation runs its loop in C.

More cores do not fix a slow algorithm. They only run the same slow steps in more places.

GIL means global interpreter lock. A thread is one worker inside your program. The GIL lets only one thread run Python code at a time.

The lock has one important gap. A thread that waits for the network or the disk lets go of the lock. Another thread can run Python code at that time.

So the GIL hurts some work and not other work. Look at the table:

The rule that follows: use threads for waiting. Use processes for computing. A process is a separate copy of Python with its own GIL, so processes run in parallel.

Python 3.13 added a free-threaded build without the GIL. A PEP is a written plan for a Python change, and this one is PEP 703.

The build was experimental in 3.13. In 3.14 it is officially supported, but it is still a separate, optional build, not the default.

Plan with the normal build, and measure before you trust any new one.

A thread pool starts a fixed number of threads and gives them tasks.

ThreadPoolExecutor, from concurrent.futures, is the standard pool. Here the slow API takes 0.5 seconds per page. time.sleep(0.5) stands in for that wait.

import time
from concurrent.futures import ThreadPoolExecutor

def fetch_page(page):
    time.sleep(0.5)  # the slow API
    return f"page {page}: ok"

pages = [1, 2, 3, 4, 5, 6]

start = time.perf_counter()
serial = [fetch_page(p) for p in pages]
print("serial:", round(time.perf_counter() - start, 1), "s")

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=6) as pool:
    threaded = list(pool.map(fetch_page, pages))
print("threads:", round(time.perf_counter() - start, 1), "s")

print(threaded == serial)
print(threaded[0], threaded[5])

The loop waits 6 times, one after another: 6 × 0.5 is about 3 seconds. With 6 threads, all 6 waits happen together, so the total is about 0.5 seconds.

pool.map gives the results in the order of the input, not the order they finish. That is why the two lists are equal. If one fetch raises an error, map raises it when you read that result.

ProcessPoolExecutor works the same way as ThreadPoolExecutor. But each worker is a separate Python process with its own GIL. So CPU work runs in parallel.

Here each worker computes a checksum for one order file. The code hashes a short value 200,000 times in a Python loop. A Python loop holds the GIL, so threads cannot help here.

import hashlib
from concurrent.futures import ProcessPoolExecutor

def checksum(order_file):
    data = f"order-file-{order_file}".encode()
    for _ in range(200_000):
        data = hashlib.sha256(data).digest()
    return order_file, data.hex()[:8]

if __name__ == "__main__":
    files = [1, 2, 3, 4, 5, 6]
    with ProcessPoolExecutor(max_workers=3) as pool:
        for order_file, digest in pool.map(checksum, files):
            print(order_file, digest)

Three workers share the six files. On a machine with at least 3 free cores, this is up to about 3 times faster than a plain loop. Fewer free cores give less.

The line if __name__ == "__main__" matters. On Windows and macOS, each worker starts a fresh Python that imports your file again.

Without the guard, each worker tries to start the pool again. The workers fail with a RuntimeError, and your program stops with BrokenProcessPool.

From Python 3.14, Linux also starts workers in a way that imports your file again. So always write the guard.

Processes have three costs. Count them before you choose:

• Pickling: Python turns each argument and result into bytes, sends them between processes, and rebuilds them. Big data makes this slow.
• Start-up: each worker is a new Python. A task that takes a few milliseconds loses more time to start-up than it gains.
• Memory: each worker has its own copy of the data it uses. Three workers can use about three times the memory.

Pickling also limits what you can send. A lambda cannot be pickled. Look at what happens:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as pool:
        future = pool.submit(lambda n: n * 2, 21)
        try:
            future.result()
        except Exception as error:
            print(type(error).__name__)

asyncio runs many waiting tasks on one thread. A task says "I am waiting now" with await. Then another task runs.

A function made with async def is a coroutine. asyncio.gather runs many coroutines and returns their results in input order. A semaphore is a counter that lets only N tasks pass at the same time.

This code fetches the same 6 pages. The semaphore lets only 2 fetches run at once:

import asyncio
import time

async def fetch_page(page, gate):
    async with gate:
        await asyncio.sleep(0.5)
        return f"page {page}: ok"

async def main():
    gate = asyncio.Semaphore(2)
    results = await asyncio.gather(*(fetch_page(p, gate) for p in range(1, 7)))
    print(results)

start = time.perf_counter()
asyncio.run(main())
print(round(time.perf_counter() - start, 1), "s")

All 6 tasks start at once, but only 2 get past async with gate. The others wait their turn. Without the semaphore, you could send thousands of calls at the API in the same moment.

Use asyncio when you have very many waits, such as thousands of calls. A task costs less memory than a thread. For a few dozen calls, a thread pool is simpler.

Threads share memory. When two threads change the same value, they can overwrite each other. This bug is a race.

A change like total += 1 has three steps: read, add, write. A thread can stop between the steps.

A barrier makes each thread wait until both threads reach it. This code uses one to force the stop, so the bug shows on every run:

import threading

total = 0
barrier = threading.Barrier(2)

def add_order():
    global total
    current = total          # read
    barrier.wait()           # both threads have read 0 now
    total = current + 1      # write

threads = [threading.Thread(target=add_order) for _ in range(2)]
for t in threads: t.start()
for t in threads: t.join()
print(total)

In real code there is no barrier. The stop happens at random times, so the bug appears only sometimes. That makes it hard to find.

There are two fixes. The first is a lock. Only one thread at a time can be inside a with lock block:

import threading

total = 0
lock = threading.Lock()

def add_orders():
    global total
    for _ in range(1000):
        with lock:
            total += 1

threads = [threading.Thread(target=add_orders) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(total)

The second fix is better: share nothing. Each worker returns its own result, and one place adds them up:

from concurrent.futures import ThreadPoolExecutor

def count_paid(chunk):
    return sum(1 for status in chunk if status == "paid")

chunks = [["paid", "paid"], ["refunded", "paid"], ["paid", "pending"]]

with ThreadPoolExecutor(max_workers=3) as pool:
    counts = list(pool.map(count_paid, chunks))

print(counts)
print(sum(counts))

A lock makes threads wait for each other, which costs speed. Prefer workers that return results. Use a lock only when state must be shared.

An API often allows only a few calls at once. If you send more, it may answer "too many requests" (HTTP 429) or block you.

A bounded pool is a pool with a fixed number of workers. That number is your limit. This code counts the calls running at the same time:

import threading
import time
from concurrent.futures import ThreadPoolExecutor

running = 0
peak = 0
guard = threading.Lock()

def call_api(page):
    global running, peak
    with guard:
        running += 1
        peak = max(peak, running)
    time.sleep(0.05)
    with guard:
        running -= 1
    return page

with ThreadPoolExecutor(max_workers=3) as pool:
    results = list(pool.map(call_api, range(1, 21)))

print(len(results), "pages, at most", peak, "calls at once")

There is a second problem. The pool keeps every task you submit in a queue in memory. With 10 million tasks, the queue itself can use all your memory.

Backpressure means the slow side makes the fast side wait. Here the workers are the slow side. The loop that submits tasks is the fast side.

A bounded semaphore stops the loop when 5 tasks are waiting or running:

import threading
import time
from concurrent.futures import ThreadPoolExecutor

waiting = threading.BoundedSemaphore(5)   # Semaphore(5) would limit the same; Bounded also catches extra release() calls
in_flight = 0
peak = 0
guard = threading.Lock()

def call_api(page):
    time.sleep(0.05)
    return page

def count(delta):
    global in_flight, peak
    with guard:
        in_flight += delta
        peak = max(peak, in_flight)

def done(_future):
    count(-1)
    waiting.release()

with ThreadPoolExecutor(max_workers=3) as pool:
    for page in range(1, 41):
        waiting.acquire()            # blocks the loop when 5 are in flight
        count(+1)
        pool.submit(call_api, page).add_done_callback(done)

print("peak in flight:", peak)

The loop pauses at waiting.acquire() until a task finishes and calls release.

The same idea works in asyncio. The semaphore from the earlier section limits calls. asyncio.Queue(maxsize=5) holds at most 5 items, and await queue.put() waits while it is full.

Now you can choose. Ask two questions: what is the task waiting on, and how many tasks are there?

This table gives the answer for common shop jobs:

Memory can overrule the table. Each process holds its own copy of the data. If one worker needs 6 GB and the machine has 8 GB, even two workers do not fit.

Then use fewer workers, or send each worker a small chunk. A smaller pool that fits in memory beats a larger pool that runs out of memory.

Everything so far makes one machine do more. That is scaling up. Scaling out means many machines work on the data, with an engine such as Spark.

A cluster costs setup time. It also moves data between machines (a shuffle). For data that fits on one machine, that makes the job slower.

Look at the two options side by side:

Leave Python for Spark when you have measured and tuned, and one machine still cannot finish. Leave it for SQL when the work is joins and group-bys on tables that already live in a warehouse.

Spark is taught in its own module. The skill you use here is the same: measure first, then pick the smallest tool that works.

Each exercise opens in the Python workspace. None of them starts a thread. They train the choice, which is what an interview asks for.

Try each one with this chapter closed. If one goes wrong, return to its section.