Writing/The Unbearable Lightness of Coding
Chapter 27 · Draft · 59 min read
Concurrent Programming: Threads, Locks and the GIL
Moore's Law Can't Save Us
For decades, we had Moore's Law: the number of transistors on a chip doubles every two years. If your code was slow, just wait — in a couple of years, processors will be fast enough anyway. But somewhere in the mid-2000s, this stopped being the case: single cores weren't getting much faster, for various reasons of heat and power — so instead, they started adding more cores to a single chip. Moore's law couldn't help us anymore: if we wanted more speed, we had to learn to parallelize our code.
The problem with parallelization is that it's not how our brain works. We think in a sequential manner: cause and effect, one thing after the other — so a program, where many things are happening at once, in an order that nobody can predict, is quite counterintuitive. Not only is it counterintuitive, but it's counterintuitive in so many different ways that we had to come up with different paradigms to deal with different aspects of it: multithreading (multiple flows of execution in a single process), multiprocessing (multiple processes), distributed computing (multiple machines) — even asynchronous programming, the black sheep of the family, which is actually older than the rest, and will get a chapter all of its own.
Before we can choose which of these paradigms to use, let's make an important distinction: I/O-bound tasks spend their time waiting for input and output: reading files, receiving packets, waiting for a database to answer; while CPU-bound tasks spend their time processing this information: doing math, or parsing it and running algorithms on it — generally, making decisions. A good way to think about it is that our thoughts are much faster than our speech: one of the first tricks you learn when you start speed reading is to stop subvocalizing — that is, pronouncing the words in your head, which slows your reading down to the speed of talking. This is exactly the difference between our CPU and our I/O: our processor can do something complicated in a few nanoseconds, while even Google takes a noticeable fraction of a second to show up on our screen, orders of magnitude slower.
This distinction is important, because given a task, you might assume that if you run it on a machine with ten cores, it should run ten times faster in parallel. This isn't true, however — and why it's not true, especially in Python, takes up most of this chapter.
Before we get to that, though, there's something else worth saying: sequential code is much simpler, so if you can help it, prefer it this way. It doesn't have to deal with race conditions, deadlocks or livelocks; its logs are in chronological order; and most importantly, we — with our limited attention span — can reason about a chronological story much more easily than about a tangle. Concurrency bugs are weird, and notoriously hard to reproduce and understand.
Still, sometimes you have no choice: a graphical user interface, for example, needs at least two threads: one to listen to the user (so the Cancel button still works while something is being downloaded), and one to do the actual work, which happens in the background; a notification pops up because something in the background was listening for new email; a web server needs to talk to multiple clients at the same time.
So, the plan is this: we'll start with threads and thread pools, and see how the GIL makes CPU-bound threads useless, and how the new, free-threaded Python finally doesn't have this problem. We'll see the limits of parallelism and Amdahl's law, and talk about processes and pickling — but then we'll get to the fun part, like race conditions, locks, conditions, queues, events, semaphores and deadlocks. We'll even see how all of these can be used to implement a thread-per-connection server, which will bring us to the next chapter.
Threads
To demonstrate such a case, let's download three URLs (python.org, cnn.com and bbc.com), and print out their sizes; first, we do this sequentially, using the Timer context manager from the Performance chapter:
>>> import requests
>>> urls = [
... 'http://python.org',
... 'http://cnn.com',
... 'http://bbc.com',
... ]
>>> def page_size(url):
... return len(requests.get(url).text)
>>> with Timer():
... for url in urls:
... print(page_size(url))
51927
5831989
669055
767.682 ms
Since each size is printed only after the previous download is complete, the total time is indeed the sum of its parts, which is around 770 milliseconds on my machine (it might be different on yours, depending on your network connection).
But now, let's try again with threads. A thread is a separate flow of execution inside a process. On Linux, it's literally called a lightweight process — it has its own registers, its own stack; but otherwise, it shares the same memory space as all the other threads in the same process. It's a bit like creating a new object in the same namespace — we don't need to copy the global variables or reload the modules, we just get a new, fresh bit of state, and a new, fresh function to run its own logic inside this state, in this world. That's what makes threads so much cheaper than processes, which the operating system keeps far away from each other.
Anyway, we use threading.Thread(target=...) to create a new thread, but it doesn't start it until we call start(), and we use join() to block until it's completed (or, if it's already completed, to return immediately). In this case, we create all the threads first, and only then start them, so as to not include the overhead in our measurement; and it's fine to join them in order, even if they finish in a different order — by the time we get to the first, it might have already finished, and the other ones might have finished as well.
>>> import threading
>>> results = []
>>> threads = []
>>> for url in urls:
... thread = threading.Thread(target=lambda: results.append(page_size(url)))
... threads.append(thread)
>>> with Timer():
... for thread in threads:
... thread.start()
... for thread in threads:
... thread.join()
206.568 ms
>>> results
[669055, 669055, 669055]
Much faster, isn't it? Around 200 milliseconds. Except, that's not actually what we wanted: if you look at the results, it's the same size, three times over. This is the late-binding bug that we discussed in the Assignment chapter: when creating the lambda, we didn't actually capture the value of url, just the name, so by the time the threads start, the loop is already over and url is left with the value of the last iteration — namely, the BBC's URL. So effectively, we've created three threads that download the same thing: not very interesting, and not very useful.
The fix is trivial — we just pass the URL as an argument, so that each thread gets its own value when it's created, and we save the results in a dictionary, to keep the order (and incidentally, to see who finished first):
>>> def save_page_size(url, results):
... results[url] = page_size(url)
>>> results = {}
>>> threads = [
... threading.Thread(target=save_page_size, args=(url, results))
... for url in urls
... ]
>>> with Timer():
... for thread in threads:
... thread.start()
... for thread in threads:
... thread.join()
261.449 ms
>>> results
{'http://python.org': 51927, 'http://bbc.com': 669055, 'http://cnn.com': 5831989}
With args=(url, results), we pass the URL (and a dictionary to store the results in) to the thread, which hands them to save_page_size when it runs. And it's interesting to see how it turns out: for me, the python.org page is the smallest, and is downloaded the quickest; then the BBC, and then CNN, whose page is almost six million characters long.
But how is it possible that all three fit within the time it takes to do just one? To understand this, we need to consider how the operating system actually switches between threads. Basically, a thread can either yield, or be interrupted.
When a thread yields, it's a voluntary act: it tells the OS "hey, I'm going to do something that might take a while, so feel free to switch me out while I'm waiting". When a thread is interrupted, it's the OS calling the shots: it tells the thread "hey, you've been running for a while, so I'm switching you out to give other threads a chance".
When threads are I/O-bound, they yield most of the time: they send a request for some packet, and while it's being fetched, the OS can switch to another thread, which might send its request. So while the first thread is waiting, the second is already sending, and by the time the first one wakes up and is ready to receive its result, the second one is already waiting. In other words, all three threads end up waiting together, so it doesn't take much longer than waiting for just one of them.
This is called concurrency: several things happening at the same time, or at least, in progress at the same time. It's not the same as parallelism, where several things actually happen at the same instant; for example, on different cores. If you have an old, single-core computer, you might have had a game that also had a chat functionality; and even if the game and the chat were happening at the same time, the computer was only capable of running one program at a time, so it was actually switching between the two every few milliseconds, and we, with our perception of time, experienced it as simultaneous. Our downloads were concurrent, because they were happening at the same time — but whether they were actually parallel or not didn't really matter, because they were mostly waiting anyway.
Thread Pools and Futures
Creating and joining threads manually is pretty clunky, so Python provides us with a more convenient interface in concurrent.futures: a thread pool, a bunch of threads that are pre-created and ready to do our bidding. All we need to do is delegate them tasks:
>>> import concurrent.futures
>>> with Timer():
... with concurrent.futures.ThreadPoolExecutor(5) as executor:
... print(list(executor.map(page_size, urls)))
[51927, 5831989, 669055]
254.078 ms
It's pretty neat: we create a ThreadPoolExecutor with some number of workers, and use its map method, which works much like the builtin map: it takes a function and iterables of arguments, and invokes that function on each argument, but does so in separate threads, using the threads in the pool. It also returns the results in the same order as the arguments, not necessarily the one in which they complete; and the with-statement takes care of shutting the pool down at the end, waiting for its threads to finish.
This way, we can also limit the number of threads: if we have five workers and ten tasks, then five of them will get going, while the other five will wait for one of the workers to become available and pick them up then. If a web server would simply open a new thread for every connection, and a thousand people would connect to it at the same moment, then it would attempt to spawn a thousand threads — and the operating system would probably have a heart attack trying to keep track of all of them. Better that some clients have to wait, than that all of them choke. When it comes to I/O bound work, we can even consider having a larger pool than the number of cores, since these threads will spend most of their time waiting anyway; but if our tasks are CPU-bound, then it's usually not a good idea to have more threads than cores, since they will only compete over them anyway.
The pool also has another nice feature: we can submit tasks individually and get back a future:
>>> executor = concurrent.futures.ThreadPoolExecutor(5)
>>> future = executor.submit(page_size, 'http://python.org')
>>> future
<Future at 0x106c75ba0 state=running>
>>> future.result()
51927
>>> future
<Future at 0x106c75ba0 state=finished returned int>
>>> future = executor.submit(page_size, 'python.org')
>>> future.result()
requests.exceptions.MissingSchema: Invalid URL 'python.org': No scheme supplied. Perhaps you meant https://python.org?
>>> future.exception()
MissingSchema("Invalid URL 'python.org': No scheme supplied. Perhaps you meant https://python.org?")
This way, we can schedule one call and get back an object that represents this future result: one that doesn't exist yet, but is being worked on. We can see the future is still running, and when we call result, it blocks, waiting for the result to become available — at which point the future is finished. If the call raised an error, result would raise it in our thread (in this case, MissingSchema, since 'python.org' is not a valid URL, and should have had http:// before it); or, we can call exception and get it instead. This way, we can submit a bunch of tasks and handle them as they complete:
>>> futures = {executor.submit(page_size, url): url for url in urls}
>>> for future in concurrent.futures.as_completed(futures):
... print(futures[future], future.result())
http://python.org 51927
http://bbc.com 669055
http://cnn.com 5831989
>>> executor.shutdown()
With as_completed, we get an iterator that returns futures as they complete; so we can start working on them as soon as they do, and the dictionary lets us map the future to the URL it's handling. And there's even concurrent.futures.wait, which blocks until all of them are done (or the first one is, or the first one to raise an error), and returns two sets: the done and the pending ones.
Before we move on, I'd like to comment on a design decision. In many tutorials, you might see threading being taught like this: by subclassing Thread and overriding its run method. This is a pattern known as the template method, and it's nice and all, but it mixes two unrelated things: the service that we want to run, and the thread that actually runs it. This often leads to bugs, like forgetting to call super().__init__():
>>> class Downloader(threading.Thread):
... def __init__(self, url):
... self.url = url
... def run(self):
... print('downloading', self.url)
>>> Downloader('http://python.org').start()
RuntimeError: thread.__init__() not called
Another bug you might see is overriding start instead of run — in which case, your code actually runs synchronously, in the caller's thread, because it's start that creates a new thread and calls run within it. So instead, I recommend writing the service as a regular class, which we can test and reuse without worrying about any threads; and then plugging it into a thread (or a pool) by passing its method as a target:
>>> class Downloader:
... def __init__(self, url):
... self.url = url
... def run(self):
... print('downloading', self.url)
>>> downloader = Downloader('http://python.org')
>>> thread = threading.Thread(target=downloader.run)
>>> thread.start(); thread.join()
downloading http://python.org
Once again, composition over inheritance.
The GIL
Of course, that's not always the case: sometimes, we have a CPU-bound task, like this deliberately naive is_prime function, which tries to divide our number by all its potential divisors, from 2 up to it, to see if any of them go in evenly. On three large enough primes, like 15485867, 15486277 and 15486727, this takes quite some time — all three are prime, after all, so it has to go all the way and try some 15 million different divisions:
>>> def is_prime(n):
... for d in range(2, n):
... if n % d == 0:
... return False
... return True
>>> numbers = 15485867, 15486277, 15486727
>>> with Timer():
... print([is_prime(number) for number in numbers])
[True, True, True]
1595.606 ms
So it takes 1.6 seconds, which is half a second per number. Let's see what happens when we do the same with a ThreadPoolExecutor:
>>> import concurrent.futures
>>> with Timer():
... with concurrent.futures.ThreadPoolExecutor(5) as executor:
... print(list(executor.map(is_prime, numbers)))
[True, True, True]
1477.520 ms
Surprise: there's no speedup at all — in fact, on some machines and versions it can even be slower than doing it sequentially, like when I first ran this code in class and it took 3.5 seconds instead of 3, because my threads were fighting each other.
The reason for this has to do with the way that CPython is implemented, which is called an interpreter. In short, it compiles your code into some intermediate representation called bytecode, which is then executed by the Python virtual machine, one instruction at a time (we actually glimpsed this process in the Functions 3 chapter). The problem is, many parts of the interpreter are not thread-safe, like the reference counts, or Python's dictionaries and memory allocator — so CPython protects them in the crudest way possible, by a single, global lock called the Global Interpreter Lock, or GIL. This lock needs to be held in order to execute bytecode, so in practice, it means that only one thread can actually be running Python code at any given moment, no matter how many cores your computer has.
Admittedly, this was done for a reason — it made CPython dramatically simpler, and single-threaded Python code considerably faster, since it only has to acquire one lock instead of thousands of fine-grained ones. Python's creator, Guido van Rossum, made this decision in favor of simplicity, and knowingly gave up the possibility of true parallelism in Python code — which is why our CPU-bound threads didn't help. Our I/O-bound threads did, because when a thread is waiting for I/O (or sleeping), it voluntarily releases the GIL, so in our first case, the first thread asked for its page and let go, then the second asked for its page and let go, then the third asked for its page and let go — and all of them waited together. But in the second case, the threads never waited, so they never let go of the GIL on their own, and just took turns. In fact, this can even be slower, since they can be contending over the lock: every few milliseconds, the waiting threads politely ask the one currently running if it would kindly let someone else have a go, and each time the context is switched there's some cost involved; and if the threads are working on different cores, they can end up invalidating each other's caches (the Graeae, the three old sisters of Greek mythology who shared a single eye between them, come to mind — Disney's Hercules gave the same gag to the Fates: the more they fight over it, the less any of them gets to see).
To illustrate, consider a simple function and its disassembly:
>>> import dis
>>> def increment():
... global counter
... counter += 1
>>> dis.dis(increment)
1 RESUME 0
3 LOAD_GLOBAL 0 (counter)
LOAD_CONST 1 (1)
BINARY_OP 13 (+=)
STORE_GLOBAL 0 (counter)
RETURN_CONST 0 (None)
>>> import sys
>>> sys.getswitchinterval()
0.005
Even a seemingly simple line, like counter += 1, is actually several instructions, like loading the global, adding to it, and storing it back. The GIL makes each of these instructions atomic, not each line of Python code — so it protects the interpreter, not your program, which is why we'll still need locks later in this chapter. As for the switch interval, it's simply how long a thread is allowed to hold the GIL while others are waiting, before it's politely asked to release it — by default, it's 5 ms.
Of course, there are ways to bypass this: C extensions can release the GIL while they do some crunching in C — this is exactly what NumPy (or hashlib, when working with large inputs) does, and what ctypes does when calling a C function. There's also a long-running effort to remove the GIL altogether, known as the gilectomy; and lastly, there are processes, each with their own interpreter and their own GIL. Incidentally, David Beazley has no less than three talks about the GIL, each one about a different, mindbending aspect of it — so take a shot (three times).
Free-Threaded Python
For years, the GIL slide in my course ended with the sentence "the gilectomy is not going so well :[". Several attempts were made to remove the GIL, like Larry Hastings's Gilectomy, but all of them made the single-threaded case much slower, and that was a non-starter, since most Python code is single-threaded. Then, Sam Gross demonstrated that it can be done, and with an acceptable overhead at that: using techniques like biased reference counting (cheap counting for the thread that owns an object), immortal objects that don't get counted at all, and small per-object locks, he has turned his idea into PEP 703, which was accepted in 2023, and shipped in Python 3.13 as an experimental, separate build: the free-threaded build, usually installed as python3.13t (with uv, it's uv python install 3.13t; if you're compiling CPython yourself, it's the configure flag --disable-gil).
To test it, I've put the prime check in a script, primes.py, which also times both the sequential and the thread pool versions, and ran it on both builds:
$ python3.13 primes.py
sequential: 1.51 s
threads: 1.50 s
$ python3.13t primes.py
sequential: 1.66 s
threads: 0.79 s
$ python3.13t -c 'import sys; print(sys._is_gil_enabled())'
False
$ PYTHON_GIL=1 python3.13t primes.py
sequential: 1.67 s
threads: 1.71 s
On the regular build, we get exactly what you'd expect: no gain from threading whatsoever (1.51 seconds for the sequential case, versus 1.50 for the threaded one). But on the free-threaded build, the same exact code runs twice as fast (0.79 seconds versus 1.66) if we use threads. In other words, we finally do get three numbers on three cores, in parallel, at the same time. Why is it twice as fast, not three times? That's because no parallelization is perfect, which is the topic of the next section (and anyway, these numbers are on my machine, so yours will vary). But also notice the price: the sequential case is actually 10% slower (1.66 seconds versus 1.51); it seems that fine-grained locking and all the extra bookkeeping it entails are not free — which is exactly the tradeoff that Guido wanted to avoid, back when he added the GIL.
Also note that sys._is_gil_enabled() tells you whether the GIL is actually off or not. The free-threaded build is able to turn it on again (using the environment variable PYTHON_GIL=1 or the flag -X gil=1), and even does so by itself (with a warning) when you import a C extension that hasn't declared that it's safe to run without the GIL — in which case we're back to 1.7 seconds. And of course, it doesn't magically make your code thread-safe: the GIL never did that, either, only the interpreter was thread-safe, and without it your code's races will just happen more often, as we shall see shortly (and anything that used to work just because of the GIL, like multiple threads sharing one iterator, is no longer guaranteed to work).
In other words, it's a new, happy ending to an old story — but it's still quite young. In Python 3.13, it's still experimental; in 3.14, it was made officially supported (PEP 779), but not the default build. And as you might have already guessed, a lot of libraries need to catch up; so before you get too excited, check whether the C extensions you depend on support it or not, and consider the implications.
Uneven Work and Amdahl's Law
So we got it twice as fast with three threads, not three times. In fact, parallel speedups are always sublinear: they never achieve the full potential of having multiple workers, and there are several reasons for that.
The most obvious one, and arguably the easiest to address, is that work doesn't usually split evenly. To see it in action, let's do a bigger job: find all the prime numbers between 20,000 and 70,000. We can do that with the find_primes function and the is_prime predicate, both defined in a little module called prime_utils.py; so, let's split it into five equal ranges and do it in parallel, with each thread reporting back when it's done:
# chunks.py
import concurrent.futures
import time
from prime_utils import find_primes
chunks = [(20_000, 30_000), (30_000, 40_000), (40_000, 50_000), (50_000, 60_000), (60_000, 70_000)]
started = time.perf_counter()
def task(start, end):
primes = find_primes(start, end)
print(f'{start}-{end}: done after {time.perf_counter() - started:.2f} s')
return primes
with concurrent.futures.ThreadPoolExecutor(5) as executor:
futures = [executor.submit(task, start, end) for start, end in chunks]
When we run it:
$ python3.13t sequential.py
found 4673 primes in 7.79 s
$ python3.13t chunks.py
20000-30000: done after 0.91 s
30000-40000: done after 1.87 s
40000-50000: done after 2.16 s
50000-60000: done after 2.61 s
60000-70000: done after 2.85 s
It takes 7.8 seconds to complete sequentially, and only 2.9 in parallel with five chunks. But if you look at when each chunk finishes, you'll notice something interesting: the first one is done after 0.9 seconds, while the last one drags on for 2.9. Even though we split the range evenly, it doesn't mean we split the work equally: our is_prime function actually takes n steps to determine if n is prime, which means that the chunks with the bigger numbers actually take longer — and the threads that finished early were sitting around and doing nothing, while the last one was grinding alone. The total time, then, is not necessarily the average of the times it takes to complete a chunk — it's actually the time it takes to complete the slowest chunk of all.
The obvious way to address this is not to split the work evenly, but to give each worker a fair shot. Rather than deciding in advance which numbers go to which thread, we can have them get the next number from a shared counter — so whoever finishes their current number first gets the next one, and generally gets more numbers the faster it is. This idea is called work stealing, and it's what real schedulers actually do (when some worker is idle, it can "steal" some tasks from a queue of a worker that is busy).
# stealing.py
import concurrent.futures
import itertools
import threading
import time
from prime_utils import is_prime
counter = itertools.count(20_000)
lock = threading.Lock()
def worker(end):
primes = []
while True:
with lock:
n = next(counter)
if n >= end:
return primes
if is_prime(n):
primes.append(n)
started = time.perf_counter()
with concurrent.futures.ThreadPoolExecutor(5) as executor:
results = list(executor.map(worker, [70_000] * 5))
print(f'found {sum(map(len, results))} primes in {time.perf_counter() - started:.2f} s')
print('per worker:', [len(primes) for primes in results])
$ python3.13t stealing.py
found 4673 primes in 2.20 s
per worker: [1324, 839, 843, 836, 831]
It takes 2.2 seconds now, and the worker counts reveal just how uneven the distribution of the work was: one worker found more than 1,300 primes (all the first, quick ones), while the others found around 840 each. On the regular build, the same script actually takes 7 seconds, which is no better than doing it sequentially — because the problem there was never about the split.
As for the lock around next(counter): we'll talk more about locks soon, but for now, suffice to say that it guarantees that only one thread is inside a with lock: block at a time. With the GIL, it doesn't actually matter, since next on an itertools.count happens to be atomic; but on the free-threaded build, sharing an iterator between threads is not guaranteed to be safe, so we better protect it. The general idea, anyway, is to protect shared state — don't rely on accidents.
Even if we could balance the load perfectly, though, we wouldn't necessarily get five times the speed. Starting threads takes time, so there's some overhead to consider — and since they compete over the same memory and caches, they actually slow each other down.
And of course, not all work can be parallelized at all: some parts of every program have to happen in a particular order, simply by virtue of how it makes sense. This is captured by Amdahl's law, which attempts to quantify just how much of a speedup we can expect when we parallelize a program. If a proportion p of the program can be parallelized, and this part can be sped up by the factor of s, then the overall speedup is 1 / ((1 - p) + p / s), because the sequential part, 1 - p, remains the same:
>>> def amdahl(p, s):
... return 1 / ((1 - p) + p / s)
>>> amdahl(0.5, 20)
1.9047619047619047
>>> amdahl(0.9, 100)
9.174311926605506
>>> amdahl(0.9, 1_000_000)
9.999910000809995
Which isn't very encouraging. For example, if we can parallelize half the program, and do so 20 times faster, the overall speedup is only 1.9. If we can parallelize 90% of the program a hundred times faster (and that's a lot of work), we get 9.17, not 90 — and even with a million cores, the limit is 10, because the 10% that remains sequential will dominate. It's simple, really: you can't make a single pile of sand higher by using more shovels.
So why do it at all? Sometimes, there's simply no other way to achieve the necessary throughput, so we have to do it anyway. But also, when we think about a system in terms of its parts — and imagine those parts running in parallel, like different services working on some shared resources like a database or a message queue — we're not only making it more concurrent, but also designing for scale, maintainability and extensibility at the same time.
Multiprocessing
So if a single interpreter can only run one thread at a time, what if we run multiple interpreters? That's the idea behind processes: separate programs, each with their own memory and their own GIL, which the operating system can schedule on different cores. That's true parallelism, even on the regular build.
To do that, we use a process pool; and to do that, we use concurrent.futures.ProcessPoolExecutor, which has the same interface as the thread pool, so it's a one-word change. In theory, at least; in practice, if you try this in the REPL, it doesn't work:
>>> import concurrent.futures
>>> def is_prime(n):
... for d in range(2, n):
... if n % d == 0:
... return False
... return True
>>> numbers = 15485867, 15486277, 15486727
>>> with concurrent.futures.ProcessPoolExecutor(3) as executor:
... print(list(executor.map(is_prime, numbers)))
Process SpawnProcess-1:
Traceback (most recent call last):
...
AttributeError: Can't get attribute 'is_prime' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.
>>> from prime_utils import is_prime
>>> with concurrent.futures.ProcessPoolExecutor(3) as executor:
... print(list(executor.map(is_prime, numbers)))
[True, True, True]
What's happening here is that processes don't share memory, so everything has to happen through inter-process communication. The function and its arguments are pickled (serialized into bytes), sent to a worker process, unpickled, run and then pickled back, sent to the main process, and finally unpickled and returned. The reason it doesn't work is because functions are pickled by reference — that is, as a module and a name, which the worker process has to import in order to find them. But on macOS and Windows, the default way to start new processes is called spawn, which means launching fresh interpreters — and since their __main__ module is a clean slate that never heard of a function we just typed into the REPL, they can't find it; but if we import it from a module, suddenly it becomes available to the worker. The reason it works on Linux is that the default way to start new processes there used to be called fork, which literally copies the parent's memory — so all the functions defined in it are copied over, and everything seems to work. Since Python 3.14, the default there is forkserver, so this is no longer the case — but the point is that it's not supposed to work anyway; in a script, the same rule would apply, and the code that creates the pool would have to be under the if __name__ == '__main__' clause, because every worker process it spawns imports the main module, and without the guard, would try to start a pool of its own.
So, in a script:
# processes.py
import concurrent.futures
import time
from prime_utils import is_prime
numbers = 15485867, 15486277, 15486727
if __name__ == '__main__':
started = time.perf_counter()
with concurrent.futures.ProcessPoolExecutor(3) as executor:
print(list(executor.map(is_prime, numbers)))
print(f'{time.perf_counter() - started:.2f} s')
$ python3.13 processes.py
[True, True, True]
0.57 s
That's 0.57 seconds instead of 1.5 — pretty crazy, considering it's the regular build, and it's almost 3 times faster.
>>> import pickle
>>> pickle.dumps(len)
b'\x80\x04\x95\x14\x00\x00\x00\x00\x00\x00\x00\x8c\x08builtins\x94\x8c\x03len\x94\x93\x94.'
>>> pickle.dumps(lambda x: x)
_pickle.PicklingError: Can't pickle <function <lambda> at 0x102aab1a0>: attribute lookup <lambda> on __main__ failed
>>> pickle.dumps(i for i in range(3))
TypeError: cannot pickle 'generator' object
What can't be pickled is anything that Python wouldn't be able to reconstruct given its name. A builtin function like len can be pickled into just a few bytes that simply say "builtins.len"; a lambda function, on the other hand, has no name to look up; and a generator is a function in the middle of its run, with a frame and a local state, which Python doesn't even attempt to transport.
# pickling.py
import concurrent.futures
def add_to_all(numbers, n):
def add(x):
return x + n
with concurrent.futures.ProcessPoolExecutor(3) as executor:
return list(executor.map(add, numbers))
if __name__ == '__main__':
print(add_to_all([1, 2, 3], 10))
$ python3.13 pickling.py
Traceback (most recent call last):
...
AttributeError: Can't get local object 'add_to_all.<locals>.add'
The same holds for nested functions, which are local objects: the worker process can't import add_to_all.<locals>.add. And generally, anything that uses a closure to keep a local state, a lambda, or an object that can't be pickled — none of those can be pickled either.
# pickling_fixed.py
import concurrent.futures
import functools
def add(x, n):
return x + n
def add_to_all(numbers, n):
with concurrent.futures.ProcessPoolExecutor(3) as executor:
return list(executor.map(functools.partial(add, n=n), numbers))
if __name__ == '__main__':
print(add_to_all([1, 2, 3], 10))
$ python3.13 pickling_fixed.py
[11, 12, 13]
The solution is to keep our worker function on the module level, and pass in whatever state we want as an argument; a functools.partial is picklable as long as its function and the arguments it binds are.
The bottom line is that processes are heavier, so they're not ideal for tiny tasks. For big chunks of work, though, they're the way to go; as an aside, when I was preparing to teach this, I was planning to do a multiprocessing echo server as well, but it turned out that what's picklable and what's not is so fundamental to its operation that it took over the entire example.
Child Processes and Distributed Computing
Besides being a way to escape the GIL, though, processes are also useful in that they let us run other programs — which is exactly what the subprocess module is for, and threads turn out to be useful around it, too. In one of my projects, for example, I needed to execute commands with a timeout: but before terminating them rudely, I wanted to politely ask them to stop, and only resort to violence if they ignored me.
For those unfamiliar with the details, terminate sends the SIGTERM signal, which is more like a request: the process can catch it and shut down gracefully, or simply ignore it altogether. kill, on the other hand, sends the SIGKILL signal, which can't be caught or ignored — the operating system simply removes the process, so it's important to give it a chance to wrap up if possible. A more considerate program would do:
>>> import shlex, subprocess, threading
>>> def execute(command, *, timeout=None, sigterm_timeout=3):
... process = subprocess.Popen(shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
... output = {}
... def communicate():
... output['stdout'], output['stderr'] = process.communicate()
... thread = threading.Thread(target=communicate)
... thread.start()
... thread.join(timeout)
... if thread.is_alive():
... process.terminate() # SIGTERM: please stop
... thread.join(sigterm_timeout)
... if thread.is_alive():
... process.kill() # SIGKILL: you're done
... thread.join()
... return process.returncode, output['stdout'], output['stderr']
>>> execute('echo hello world')
(0, b'hello world\n', b'')
>>> with Timer():
... print(execute('sleep 10', timeout=1))
(-15, b'', b'')
1008.717 ms
>>> with Timer():
... print(execute('''sh -c 'trap "" TERM; exec sleep 10' ''', timeout=1, sigterm_timeout=1))
(-9, b'', b'')
2017.878 ms
Here, we use Popen to start a command (capturing its outputs), and communicate to read both of them to the end and wait for the process to finish — but we do so in a thread, and wait for it (in the main thread) with join(timeout), which returns after at most timeout seconds, whether the thread is done or not, so we double-check with is_alive(). If it is, we timed out, so we terminate the process, give it sigterm_timeout seconds to finish up, and kill it outright if it doesn't. Note that the arguments to execute are keyword-only (see the Functions 1 chapter), which I find a nice touch: both timeout and sigterm_timeout are configuration, and forcing the caller to name them explicitly helps keep the code readable. And indeed, it works.
So echo returns exit code 0 and its output; sleep 10 is terminated after 1 second, and its exit code is -15, meaning "killed by signal 15", or SIGTERM. If we try to ignore SIGTERM (with the trap "" TERM; exec sleep 10 trick, which makes the shell ignore it and then replaces itself with sleep, thus "inheriting" the signal behavior), it does survive the termination — but is killed a moment later, and returns -9, for SIGKILL.
Today, this is more or less subprocess.run(command, timeout=...), except it kills the child process immediately with SIGKILL (and then raises a TimeoutExpired exception), skipping this polite step entirely. If you're running programs that need a chance to clean up, though, something like execute might still be useful.
But eventually, even all the cores of the biggest machine you can buy are not enough, and you scale horizontally, adding more machines rather than making a single one bigger. The analogy is simple: when you're scaling vertically, you make a single pile higher — but at some point, a pile can only be so high (just try buying a computer with a petabyte of RAM). With horizontal scaling, you add more piles — and you can always buy a thousand computers with a terabyte each. That's how Google built its early infrastructure — using a technique called MapReduce and its own file system, it was able to run everything on a large number of regular machines.
Of course, this comes with its own overhead — and it's a depressing one, because it's entirely unavoidable: network traffic, cables, protocols, serialization, machines failing independently — and even the CAP theorem, which states that when the network partitions, you must choose between consistency and availability, which is why systems like these so often settle for eventual consistency. But it works: throw enough money on the problem — more threads, processes, entire hosts — and it goes away; the only problem is that it isn't very respectful towards the environment, seeing as most of these machines spend most of their time waiting. Keep this thought in the back of your head; it's where our next chapter begins.
Race Conditions
Now, it's important to understand what makes threads so dangerous, because they are quite seductive. You see, they share memory, which is what makes them so light; but it's also what makes them so volatile. To illustrate, suppose that we have a simple server, which receives a message and saves it to a file, messages.txt, by reading it, appending the message to it, and writing it back; and consider what happens when two threads call this function at the same time, trying to save two different messages, 'hello' and 'hi':
>>> import threading, time
>>> open('messages.txt', 'w').close()
>>> def save(message):
... with open('messages.txt') as f:
... messages = f.read()
... messages += message + '\n'
... time.sleep(0.01) # a context switch, at the worst possible moment
... with open('messages.txt', 'w') as f:
... f.write(messages)
>>> t1 = threading.Thread(target=save, args=('hello',))
>>> t2 = threading.Thread(target=save, args=('hi',))
>>> t1.start(); time.sleep(0.005); t2.start()
>>> t1.join(); t2.join()
>>> print(open('messages.txt').read())
hi
Let's follow the code, and see if we can understand what's going on. First, thread 1 reads the empty file, and adds "hello" to it. But before it can write it back, the operating system switches to thread 2 (the sleep just makes this unlucky moment certain), who also reads the empty file, and adds "hi" to it. Now, thread 1 wakes up and writes its "hello" to disk, overwriting the original, empty version; and thread 2, none the wiser, writes its "hi" over it. As a result, we lose the "hello" altogether. This is known as a race condition, because the final result depends on the "race" between the threads — which one happens to read the file first, and which one happens to write it last. Note that the interleaving is up to the operating system, which knows nothing about our program's logic: the code is correct on most runs, and only wrong sometimes, which is what makes these bugs so notoriously hard to reproduce and debug.
Let's see it in action, with a hundred threads; this time, we don't even bother with an artificial sleep:
>>> def save(message):
... with open('messages.txt') as f:
... messages = f.read()
... messages += message + '\n'
... with open('messages.txt', 'w') as f:
... f.write(messages)
>>> open('messages.txt', 'w').close()
>>> threads = [threading.Thread(target=save, args=(f'message {i}',)) for i in range(100)]
>>> for thread in threads:
... thread.start()
>>> for thread in threads:
... thread.join()
>>> len(open('messages.txt').read().splitlines())
3
If we run this code, we'll get a hundred messages in, but only three lines out (although the exact number seems to fluctuate between runs), because the operating system switches between the threads all the time, and file operations release the GIL, so they actually interleave in the middle, when some threads are still reading, and some are already writing.
So, what's the problem? It's the part of the code that reads, modifies and writes the file, known as the critical section, because it's the bit that deals with the shared resource; and it needs to be mutually exclusive, so that only one thread can be in it at any given moment.
But doesn't the GIL take care of this? Isn't the whole point of having it that it makes Python code threadsafe, by preventing such race conditions? The answer is no — it doesn't, and it's not.
Think about it this way: the GIL makes single bytecode instructions atomic, not sequences of them. counter += 1 is not an instruction, it's three: load the value, add 1 to it, and store it back. Even if these instructions are carried out one by one, without interruption, it still doesn't help: a thread can load the value, then be switched out, and when it resumes, it will add 1 to that value, and store it back — but in the meantime, the value could have already been updated by other threads, and this particular update would be lost. To prove this, we can write counter.py:
# counter.py
import threading
counter = 0
def increment(times):
global counter
for _ in range(times):
counter += 1
threads = [threading.Thread(target=increment, args=(1_000_000,)) for _ in range(4)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
print(f'{counter:,}')
$ python3.13 counter.py
4,000,000
$ python3.13t counter.py
1,052,514
On regular 3.13, this prints 4000000 — but don't be fooled. Recent versions of CPython only switch threads at certain points, like when they jump back to the top of a loop; so this particular race condition doesn't happen to manifest. In older versions, though, and in different scenarios, it absolutely did; and in any case, nothing in the language promises otherwise. On the free-threaded build, the threads actually do run at once; and in this case, almost three quarters of the increments are lost. So the GIL wasn't solving the problem — it was simply hiding it.
For completeness sake, I should mention that there are other concurrency bugs, too: like deadlocks, when threads end up waiting for each other indefinitely (which we'll encounter later in this chapter), or starvation, when a thread never gets its chance. There are even livelocks, when threads are constantly reacting to each other and getting in each other's way, like two people who want to pass each other in a narrow corridor, and keep stepping aside in the same direction at the same time.
Locks
The cure is simple, and it's called a lock, also known as a mutex or a mutual exclusion object; we actually met it briefly in the Objects 5 chapter. It's an object that can be held by one thread at a time: and since acquiring it is an atomic operation, a single, indivisible step, if two threads try to do it at the same time, one of them will get it, while the other will be blocked until it's released.
The analogy I like here is a talking stick: if a bunch of people sit in a circle, and anyone is allowed to talk as long as they hold a stick, then we can emulate this with a mutex: only the person holding the stick will be allowed to speak.
In Python, we can achieve the same like so:
>>> lock = threading.Lock()
>>> def save(message):
... with lock:
... with open('messages.txt') as f:
... messages = f.read()
... messages += message + '\n'
... with open('messages.txt', 'w') as f:
... f.write(messages)
>>> open('messages.txt', 'w').close()
>>> threads = [threading.Thread(target=save, args=(f'message {i}',)) for i in range(100)]
>>> for thread in threads:
... thread.start()
>>> for thread in threads:
... thread.join()
>>> len(open('messages.txt').read().splitlines())
100
All 100 messages are there; and note that the with statement is important — if we just acquired and released the lock manually, then if the critical section raised an exception and nobody released the lock, all the other threads would be waiting for it indefinitely, and the program would hang. So we never want to acquire and release a lock manually, when a with statement can do so in a try-finally clause for us.
The first time I taught this, I actually made a mistake that I had to correct in the following lesson, and it's quite instructive. You see, my Handler class created its lock as part of its constructor:
>>> class Handler:
... def __init__(self, message):
... self.message = message
... self.lock = threading.Lock()
... def run(self):
... with self.lock:
... with open('messages.txt') as f:
... messages = f.read()
... messages += self.message + '\n'
... with open('messages.txt', 'w') as f:
... f.write(messages)
>>> open('messages.txt', 'w').close()
>>> handlers = [Handler(f'message {i}') for i in range(100)]
>>> threads = [threading.Thread(target=handler.run) for handler in handlers]
>>> for thread in threads:
... thread.start()
>>> for thread in threads:
... thread.join()
>>> len(open('messages.txt').read().splitlines())
15
That's 15 out of 100 lines. Each Handler locked and unlocked its own lock, and wasn't really looking at anybody else — because really, a lock is only useful when there's one of it and everyone is synchronizing on that one. As was rightly pointed out to me, it actually synchronized nothing; so the class needed to be changed to:
>>> class Handler:
... lock = threading.Lock()
... def __init__(self, message):
... self.message = message
... def run(self):
... with self.lock:
... with open('messages.txt') as f:
... messages = f.read()
... messages += self.message + '\n'
... with open('messages.txt', 'w') as f:
... f.write(messages)
>>> threads = [threading.Thread(target=Handler(f'message {i}').run) for i in range(100)]
>>> for thread in threads:
... thread.start()
>>> for thread in threads:
... thread.join()
>>> len(open('messages.txt').read().splitlines())
100
Since the lock is now a class attribute, it's the same object across all the instances — as we've seen in the Objects chapters — so they actually do wait for each other; and the corollary is that an anonymous lock, like with threading.Lock():, is always pointless.
As for granularity, having one lock per class — like we did with the @threadsafe class decorator in the Metaclasses chapter — is coarse, but simple; while having a lock per instance (or even per file) lets unrelated work happen in parallel, but is easier to get wrong. If performance is not critical, it's best to start coarse; if it's too slow, we can always optimize it later.
>>> import threading
>>> lock = threading.Lock()
>>> lock.acquire()
True
>>> lock.acquire(timeout=1)
False
>>> lock.release()
>>> rlock = threading.RLock()
>>> rlock.acquire()
True
>>> rlock.acquire()
True
>>> rlock
<locked _thread.RLock object owner=8440226176 count=2 at 0x10393a700>
>>> rlock.release(); rlock.release()
Note that a plain Lock doesn't actually know who's holding it, so a thread trying to acquire it while it's already holding it will end up waiting for itself forever. Adding a timeout at least lets it give up after a second, and return False instead of waiting forever; an RLock, or reentrant lock, on the other hand, remembers who its owner is, and can be acquired multiple times by the same thread (which might be useful when one method calls another) — although in this case we need to release it an equal number of times to actually unlock it.
One thing to note, though, is that a threading.Lock only works inside a single process: if you want to synchronize several processes, or several machines, you'd have to use a lock at the network level. One popular solution for that is to use a key in a shared store, like Redis; and the reason its commands are so odd is actually very sensible. The INCR operation is an atomic increment — that is, it increments the value and returns the result in a single, indivisible step — while GETSET returns the value and writes a new one in one; so GETSET "locked" returns "unlocked" to one and only one client, who is now in possession of the lock.
The reason we can't just read it, and if it's "unlocked" then set it to "locked" and proceed, is that this is a race condition: two clients can read it as "unlocked" at the same time, and both set it to "locked" and proceed to their critical section, assuming that only one of them is there. It's called a check-then-act race, and it's the most common race condition of all, and the reason atomic operations that can do both at the same time, like GETSET, even exist.
As an aside, these days, Redis actually recommends using a different command to achieve the same — namely, SET with the NX (for not-exists) option and an expiry — but if you're interested in the subject, there's a whole, beautiful theory of concurrent data structures, many of which are lock-free.
Conditions and Queues
A classic way of using threads is to implement a message queue: one or more threads can publish a message, and one or more threads can subscribe to receive these messages. Here's a simple implementation that I used to show in class — it uses a lock, but see if you can figure out what's wrong with it before reading on:
>>> class MessageQueue:
... def __init__(self):
... self.messages = []
... self.subscribers = []
... self.lock = threading.Lock()
... def publish(self, message):
... with self.lock:
... self.messages.append(message)
... def subscribe(self, function):
... self.subscribers.append(function)
... return function
... def run(self):
... while True:
... if not self.messages:
... continue
... message = self.messages.pop(0)
... for subscriber in self.subscribers:
... subscriber(message)
>>> mq = MessageQueue()
>>> @mq.subscribe
... def f1(message):
... print(f'1: {message}')
>>> @mq.subscribe
... def f2(message):
... print(f'2: {message}')
>>> threading.Thread(target=mq.run, daemon=True).start()
>>> mq.publish('Hello, world!')
1: Hello, world!
2: Hello, world!
It works, in that we can publish from the main thread, and both subscribers will receive and print the message; but it's actually wrong on two accounts.
First, in run, we check the list, pop something off of it, and iterate over the subscribers, all without acquiring the lock. This was actually pointed out to me by a student during one of the classes, and he was right: it happens to work, but only because of the GIL, which guarantees that append and pop are each a single operation. But that doesn't mean it's correct: checking the list and then popping from it is a check-then-act race condition, and it breaks as soon as you have two consumers, in which case they might both see one message, and both try to pop it off.
Second, it busy-waits: while True: if not self.messages: continue keeps asking "are we there yet?" as fast as the CPU allows. Let's measure it:
>>> import resource
>>> cpu = lambda: resource.getrusage(resource.RUSAGE_SELF).ru_utime
>>> before = cpu(); time.sleep(1); cpu() - before
0.9781419999999998
That's almost a full second of CPU time for a process that actually spent one second doing nothing. That's an entire core, burnt on asking whether it's time to do something yet. You could sleep for a second in the loop, and it would help; but then every message will take up to a second to be delivered. What we actually want is for the consumer to sleep until something actually changes.
That's what condition variables are for: a threading.Condition object is associated with a lock; calling its wait method releases that lock, and goes to sleep, in one atomic step, until another thread calls notify (or notify_all), at which point it re-acquires the lock and continues. wait_for(predicate) does this in a loop, until the predicate returns True, which is important — because a thread that wakes up should check again, in case another consumer beat it to the message. The queue can be rewritten like so:
>>> class MessageQueue:
... def __init__(self):
... self.messages = []
... self.subscribers = []
... self.condition = threading.Condition()
... def publish(self, message):
... with self.condition:
... self.messages.append(message)
... self.condition.notify()
... def subscribe(self, function):
... with self.condition:
... self.subscribers.append(function)
... return function
... def run(self):
... while True:
... with self.condition:
... self.condition.wait_for(lambda: self.messages)
... message = self.messages.pop(0)
... subscribers = list(self.subscribers)
... for subscriber in subscribers:
... subscriber(message)
>>> threading.Thread(target=mq.run, daemon=True).start()
>>> mq.publish('Hello, world!')
1: Hello, world!
2: Hello, world!
>>> before = cpu(); time.sleep(1); cpu() - before
4.599999999999743e-05
Publishing appends a message and notifies any potential waiters, while running waits for a message, pops it, and copies the subscribers list, all while holding the lock — but then calls the subscribers outside of the lock, since calling unknown code while holding a lock is usually a bad idea; if a subscriber were to publish a message, it would need to acquire the very lock we're holding. This time, we don't burn an entire core on asking "are we there yet?" — that's 46 microseconds, down from a whole second; not bad.
You probably won't have to implement this yourself, though, since Python comes with queue.Queue: a thread-safe FIFO queue, implemented in exactly the same way, with a lock and a couple of conditions inside:
>>> import queue
>>> class MessageQueue:
... def __init__(self):
... self.messages = queue.Queue()
... self.subscribers = []
... def publish(self, message):
... self.messages.put(message)
... def subscribe(self, function):
... self.subscribers.append(function)
... return function
... def run(self):
... while (message := self.messages.get()) is not None:
... for subscriber in self.subscribers:
... subscriber(message)
... def stop(self):
... self.messages.put(None)
>>> thread = threading.Thread(target=mq.run)
>>> thread.start()
>>> mq.publish('Hello, world!')
1: Hello, world!
2: Hello, world!
>>> mq.publish('Goodbye, world!')
1: Goodbye, world!
2: Goodbye, world!
>>> mq.stop(); thread.join()
>>> thread.is_alive()
False
Publishing just adds an item, consuming just gets an item (and blocks if there isn't any), and there's no lock or busy loop in sight. Also, isn't that walrus operator nice here?
To stop the consumer, we put a sentinel (None), which also lets us use a regular, non-daemon thread and join it in a clean way. One caveat, though: if we were to subscribe to messages while the queue is running, we'd also need a lock to protect the subscribers list; but here, we subscribe before starting it.
This queue also has a few other interesting features:
>>> import queue
>>> q = queue.Queue(maxsize=2)
>>> q.put(1); q.put(2)
>>> q.put(3, timeout=1)
queue.Full
>>> q.get(), q.get()
(1, 2)
>>> q.get(timeout=1)
_queue.Empty
For example, if we set a maxsize, put will block if the queue is full, which gives us backpressure: fast producers are slowed down to the speed of the consumers, instead of filling up the memory. Both put and get also accept a timeout, and raise queue.Full or queue.Empty if it expires; and there's also task_done and join, if we want to wait until every item has been processed, as well as LifoQueue and PriorityQueue if we want different orders.
Events, Semaphores and Deadlocks
Two other primitives that we might encounter in concurrent programming are the Event and the Semaphore objects, so let's cover them briefly as well.
An event is simply a flag that multiple threads can wait for. It has set, clear, is_set and wait methods — so threads can check it, or wait until it becomes true. Here's a thread that uses an event to loop until the party is over:
>>> stop = threading.Event()
>>> def heartbeat():
... while not stop.is_set():
... print('beep')
... stop.wait(0.3)
... print('stopped')
>>> thread = threading.Thread(target=heartbeat)
>>> thread.start(); time.sleep(1)
beep
beep
beep
beep
>>> stop.set(); thread.join()
stopped
Note that because the thread is napping with stop.wait(0.3) instead of time.sleep(0.3), we can stop it immediately when the event is set by calling stop.set(), and don't have to wait for it to finish its nap. This is the standard way to stop a thread in Python, since there's no way to kill a thread from outside — so threads have to be asked to stop, and need to check this regularly.
A semaphore is a counter of permits: it has an acquire method, which attempts to get one, blocking if it's not available; and a release method, which gives one back. In other words, a lock is a semaphore with just one permit — but you can also have a semaphore with n permits, allowing up to n threads to access a section at the same time. This is useful, for example, when we want to limit access to some resource, like a server that can handle five connections at once, or an API that only allows two requests at a time. Here's how six downloads might be simulated in six threads, but limited to two at a time:
>>> slots = threading.Semaphore(2)
>>> def download(url):
... with slots:
... print(f'{time.perf_counter() - started:.1f} s: downloading {url}')
... time.sleep(0.5)
>>> urls = [f'http://example.com/{i}' for i in range(6)]
>>> started = time.perf_counter()
>>> with concurrent.futures.ThreadPoolExecutor(6) as executor:
... results = executor.map(download, urls)
0.0 s: downloading http://example.com/0
0.0 s: downloading http://example.com/1
0.5 s: downloading http://example.com/2
0.5 s: downloading http://example.com/3
1.0 s: downloading http://example.com/4
1.0 s: downloading http://example.com/5
That's why we see just two downloads happening at any given moment: two at 0.0 seconds, two at 0.5 seconds and two at 1.0 seconds. A BoundedSemaphore is the same, but raises an error if it's released more times than it was acquired — which usually indicates a bug.
At this point, we're ready to talk about the other classic concurrency bug: deadlocks.
Imagine two bank accounts, each with their own lock, which is fine-grained enough to allow transfers between unrelated accounts in parallel. However, if account A tries to transfer money to account B, and account B tries to transfer money to account A at the same time, we might end up with:
>>> class Account:
... def __init__(self, name, balance):
... self.name = name
... self.balance = balance
... self.lock = threading.Lock()
>>> def transfer(source, target, amount):
... with source.lock:
... time.sleep(0.1) # a context switch, at the worst possible moment
... with target.lock:
... source.balance -= amount
... target.balance += amount
>>> alice, bob = Account('alice', 100), Account('bob', 100)
>>> t1 = threading.Thread(target=transfer, args=(alice, bob, 10), daemon=True)
>>> t2 = threading.Thread(target=transfer, args=(bob, alice, 20), daemon=True)
>>> t1.start(); t2.start()
>>> t1.join(timeout=2); t2.join(timeout=2)
>>> t1.is_alive(), t2.is_alive()
(True, True)
Thread 1 attempts to transfer money from alice to bob; it acquires alice's lock, and then (I added the sleep to make the unfortunate timing certain) attempts to acquire bob's lock. Thread 2, meanwhile, attempts to transfer money from bob to alice; it similarly acquires bob's lock, and then attempts to acquire alice's lock. As a result, each thread holds the lock the other wants, and they end up in a deadlock: both waiting for each other forever. The joins time out after two seconds, so both threads are still alive (I made them daemon threads, otherwise the program would just hang). A deadlock requires four conditions: mutual exclusion, where some threads hold a lock that others want; a hold-and-wait situation, where a thread holds one lock while waiting for another; no preemption, where nobody can just take away a lock; and a circular wait. Breaking any of these conditions can make deadlocks impossible, and the easiest one to break is the circle:
>>> def transfer(source, target, amount):
... first, second = sorted([source, target], key=lambda account: account.name)
... with first.lock:
... time.sleep(0.1)
... with second.lock:
... source.balance -= amount
... target.balance += amount
>>> alice, bob = Account('alice', 100), Account('bob', 100)
>>> t1 = threading.Thread(target=transfer, args=(alice, bob, 10))
>>> t2 = threading.Thread(target=transfer, args=(bob, alice, 20))
>>> t1.start(); t2.start(); t1.join(); t2.join()
>>> alice.balance, bob.balance
(110, 90)
If all threads agree on a global order to acquire locks, they will all go for the same lock — in this case, alice's lock — first; so one of them will get it, and the other will wait for it to become available before even attempting to get anything else. This way, the circular wait is impossible, and everything works: the balances are 110 and 90, as expected. Of course, it doesn't have to be alphabetical order; any consistent order will do, like account numbers.
Other ways to avoid deadlocks include minimizing the number of locks you hold at once, and keeping them for as short as possible; never calling unknown code while holding a lock; using acquire with a timeout, so that a deadlock will result in an exception rather than a silent hang; and generally, trying to use higher-level constructs, like queues, that do the necessary synchronization for us.
Thread per Connection, and the Road Ahead
Let's go back to where we left off: the echo server, which receives a message and sends it back to the client. Here it is again, with the walrus in its loop, and this time started in a background daemon thread, so we can play the clients in the same REPL:
>>> class EchoServer:
... def __init__(self, host, port):
... self.host = host
... self.port = port
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... with listener:
... while True:
... connection, address = listener.accept()
... self.handle(connection)
... def handle(self, connection):
... with connection:
... while data := connection.recv(4096):
... connection.sendall(data)
>>> server = EchoServer('127.0.0.1', 8000)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8000))
>>> b = socket.create_connection(('127.0.0.1', 8000))
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b.sendall(b'world'); b.settimeout(1); b.recv(4096)
TimeoutError: timed out
>>> a.close()
>>> b.recv(4096)
b'world'
Client b connects alright (its connection is merely queued in the listener's backlog, which is what the argument to listen determines: the number of pending connections we're willing to keep in our pocket), but its echo never comes. The reason, of course, is that the server is stuck in handle, serving a — it's not in accept, and can't be in both places at once. Only when a closes its connection does the server get a chance to go back to accept, and attend to b — at which point, its "world" is finally echoed back.
To fix this, we can simply add a thread per connection — so that once it's accepted, its handling can be delegated to a new thread and the accept loop can move on to the next connection immediately. The ThreadedEchoServer is actually quite elegant:
>>> class ThreadedEchoServer(EchoServer):
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... with listener:
... while True:
... connection, address = listener.accept()
... thread = threading.Thread(target=self.handle, args=(connection,))
... thread.start()
>>> server = ThreadedEchoServer('127.0.0.1', 8001)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> a = socket.create_connection(('127.0.0.1', 8001))
>>> b = socket.create_connection(('127.0.0.1', 8001))
>>> a.sendall(b'hello'); a.recv(4096)
b'hello'
>>> b.sendall(b'world'); b.recv(4096)
b'world'
>>> threading.active_count()
4
And indeed, both clients are served at once, and threading.active_count() returns 4: the main thread, the server's thread and a thread for each client. Note that I didn't change handle at all; the threading aspect is entirely in how it's called, isn't it? (And please don't confuse the two: the thread is not the same as the service.)
The problem with this, though, is that if we have ten thousand clients, it means we'll have ten thousand threads — and each will have its own stack, which the operating system will have to keep track of. This can become a problem, so we might want to cap the number of threads with a pool:
>>> class PooledEchoServer(EchoServer):
... def start(self):
... listener = socket.socket()
... listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
... listener.bind((self.host, self.port))
... listener.listen(5)
... with listener, concurrent.futures.ThreadPoolExecutor(2) as executor:
... while True:
... connection, address = listener.accept()
... executor.submit(self.handle, connection)
>>> server = PooledEchoServer('127.0.0.1', 8002)
>>> threading.Thread(target=server.start, daemon=True).start()
>>> clients = [socket.create_connection(('127.0.0.1', 8002)) for _ in range(3)]
>>> for i, client in enumerate(clients):
... client.settimeout(1)
... client.sendall(f'client {i}'.encode())
... try:
... print(client.recv(4096))
... except TimeoutError:
... print(f'client {i} is still waiting')
b'client 0'
b'client 1'
client 2 is still waiting
With only two workers, the third client is accepted, but has to wait until one of them becomes available — that's the tradeoff, some clients waiting instead of all of them choking. (This was, incidentally, the first exercise of my systems course: a thread per connection, and a test checking for a race condition when two clients were writing to the same file at once.)
But let's take a step back, and ask a more fundamental question: what exactly are all these threads doing? Almost nothing: they just sit there, waiting for their client to say something. In other words, they're stuck in recv, waiting for something to happen — and given how slowly humans type, and how slowly even networks operate, it's quite an eternity for a CPU. We pay a price for having all these threads — namely, their stacks, context switches and locks to protect their shared state — just to wait.
The takeaway from this is that parallelism is a tool that helps us do more in less time, but it's counterintuitive in at least three ways. The GIL makes it so threads can't actually run Python code in parallel (at least, until free-threading, which finally lets them); Amdahl's law tells us that even with any amount of parallelism, we can only do so much; and if we have any shared state, we're going to have race conditions and deadlocks, unless we synchronize access carefully. So unless you have a very good reason not to, prefer the highest level of abstraction that works: use a pool and map if you can, or at the very least a queue — or even a process pool, if it's really CPU-bound.
But there's a different way of thinking about this altogether. If our threads are mostly waiting, maybe we don't need more threads to begin with — maybe we can just have one smart thread that, when it's waiting on one client, goes ahead and works on another's problem while it's waiting, too. Picture a bank: one British clerk serving a polite queue of people, one at a time; twenty British clerks serving twenty such queues; or one Israeli clerk, juggling all the customers at once. This is called asynchronous programming, and it's the topic of the next chapter.