Threads and Processes Checkpoint

30 min · Intermediate

Fix the mistakes

The slips that cost marks in this module. Spot each one, then check the fix.

This checkpoint tests the whole module: threads and locks, processes and pools, executors and the GIL. Read each question, then run the code in your head.

Pick the tool first

A report job loads 20 customer records from a slow service, then runs a heavy calculation on each. The loading waits, so it fits threads. The calculation uses the CPU, so it fits processes.

Mixing is common: a thread pool for the waiting part feeds a process pool for the calculating part.

Keep shared state safe

Threads share memory, so a shared total needs a lock. Processes do not share memory, so results must come back as return values.

In both cases, sort or key results before printing.

Example 1 · Threads for loading, one place to combine

Python

1import time2from concurrent.futures import ThreadPoolExecutor3
4def load(customer_id):5    time.sleep(0.05)               # a slow service6    return customer_id, customer_id * 1007
8with ThreadPoolExecutor(4) as ex:9    balances = dict(ex.map(load, range(1, 6)))10print(balances)11print(sum(balances.values()))

Output

{1: 100, 2: 200, 3: 300, 4: 400, 5: 500}
1500

What this code does

The five loads overlap, so the wait is about one load, not five.
map returns pairs in input order, and dict(...) builds the lookup.
Nothing is shared between jobs: each returns its own pair.

Example 2 · Processes for the calculation

Python

1from concurrent.futures import ProcessPoolExecutor2
3def risk_score(balance):4    return sum(i % 7 for i in range(balance))5
6if __name__ == "__main__":7    balances = [100, 200, 300, 400, 500]8    with ProcessPoolExecutor(2) as ex:9        scores = list(ex.map(risk_score, balances))10    print(scores)

Output

[295, 594, 897, 1197, 1494]

What this code does

The function is at module level, so it can be pickled and sent to a worker.
The result list lines up with balances.

If you find yourself locking a lot, return values from jobs instead and combine them in one place.

Finish up

Mark complete & continue

Next: async, await and the Event Loop

1import time2from concurrent.futures import ThreadPoolExecutor3 4def load(customer_id):5 time.sleep(0.05) # a slow service6 return customer_id, customer_id * 1007 8with ThreadPoolExecutor(4) as ex:9 balances = dict(ex.map(load, range(1, 6)))10print(balances)11print(sum(balances.values()))

1from concurrent.futures import ProcessPoolExecutor2 3def risk_score(balance):4 return sum(i % 7 for i in range(balance))5 6if __name__ == "__main__":7 balances = [100, 200, 300, 400, 500]8 with ProcessPoolExecutor(2) as ex:9 scores = list(ex.map(risk_score, balances))10 print(scores)