Only three burners: synchronization tools

Night four, nine-forty. There's a group booking, so three more cooks have come in for the day. Six cooks, three burners.

Minho is first to a burner and lights it. "If we warm them up early, we'll be faster." Without a word, the chef shuts the gas valve. "No flames before opening. And if all six of you crowd around the burners, nobody can move."

Sua raises her hand. "What if we hang just three number tags by the burners? Only someone holding a tag gets to stand at the flame." For the first time tonight, the chef smiles. "That's tonight's first tool."

4-1. Three burners: Semaphore (12 min)

The Lock from Lesson 3 was one piece of chalk: one person at a time. A Semaphore(n) is n number tags. Up to n people can be in at once, and person n+1 waits until someone hands a tag back.

import threading
import time

stoves = threading.Semaphore(3)         # three burners
start = time.perf_counter()

def cook(name):
    with stoves:                        # wait for a free burner, then take it
        print(f"{time.perf_counter() - start:.1f}s {name} on")
        time.sleep(0.3)

names = [f"cook{i}" for i in range(1, 7)]
threads = [threading.Thread(target=cook, args=(n,)) for n in names]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(f"total {time.perf_counter() - start:.2f}s")
# e.g.:
# 0.0s cook1 on
# 0.0s cook2 on
# 0.0s cook3 on
# 0.3s cook4 on
# 0.3s cook5 on
# 0.3s cook6 on
# total 0.60s

• Entering with stoves: takes one tag (acquire), and leaving hands it back (release). When there are zero tags left, whoever is trying to get in waits.
• Three at 0.0 seconds, three at 0.3. With three burners, six orders go in two rounds and take 0.6 seconds. Who gets a burner first may differ from run to run.
• Semaphore(1) lets one person in at a time, like a Lock. But where releasing an already released Lock raises RuntimeError, a semaphore quietly gains an extra tag if you call release() too many times (see BoundedSemaphore below).
• In real code, you use it to cap the number of resources that suffer when everyone piles on at once: "at most five connections at a time," "at most three downloads at a time."

Change the number of burners on the timeline below. Yellow bars are time spent waiting for a burner; orange bars are time actually spent cooking.

• 3 burners: three rows go straight to orange, and the other three wait in yellow for 0.3 seconds before turning orange. About 0.6 seconds in all.
• 1 burner: six cooks or not, they cook one at a time. The last cook waits 1.5 seconds, and the total is about 1.8 seconds, the same as working alone.
• 6 burners: almost no waiting, about 0.3 seconds. Once there are as many burners as cooks, it doesn't get any faster.
• Add cooks without adding burners and all you get is more yellow bars, one per extra person. The bottleneck is the number of burners, not the number of people.

BoundedSemaphore: catching the extra tag

A Semaphore doesn't count how many tags come back. If a bug calls release() one time too many, there are quietly four tags. Three burners, and four people at the flames.

loose = threading.Semaphore(3)
loose.release()                         # released one extra time by mistake
print(sum(loose.acquire(blocking=False) for _ in range(10)))   # now there are four burners
# 4

strict = threading.BoundedSemaphore(3)
strict.release()
# ValueError: Semaphore released too many times

A BoundedSemaphore raises ValueError if it gets back more than it started with. If you're using a semaphore to cap a count, BoundedSemaphore is almost always the better choice. Mistakes blow up on the spot instead of slipping by.

4-2. "Open!": Event (9 min)

An Event is a single signal light on the kitchen wall. It starts off, and when someone turns it on (set()), everyone who was waiting for it (wait()) moves all at once.

opened = threading.Event()              # starts off
start = time.perf_counter()

def cook(name):
    opened.wait()                       # wait until the signal comes on
    print(f"{time.perf_counter() - start:.1f}s {name} fire")

threads = [threading.Thread(target=cook, args=(n,)) for n in ["minho", "sua", "tak"]]
for th in threads:
    th.start()
time.sleep(0.5)
print(opened.is_set())
opened.set()                            # "Open!"
for th in threads:
    th.join()
print(opened.is_set())
# False
# 0.5s minho fire     (the order of these three lines varies by run)
# 0.5s sua fire
# 0.5s tak fire
# True

You can also give wait() a time limit. Then the return value matters: True if the signal came on, False if time ran out and it's still off.

closing = threading.Event()
print(closing.wait(timeout=0.2))        # waited 0.2 s, no signal
closing.set()
print(closing.wait(timeout=0.2))        # already on, so True right away
closing.clear()                         # turn it off again
print(closing.is_set())
# False
# True
# False

One pattern comes up a lot: make a "closing time" signal with an Event, and have each worker thread check if closing.is_set(): break on every pass of its loop. Unlike the daemon threads from Lesson 2, which get cut off midway, this lets threads clean up and finish on their own.

The timeline below is a kitchen where the chef shouts "Open!" after 0.5 seconds.

• Every row waits in yellow from 0 to 0.5 seconds, then turns orange all at once at 0.5. About 0.8 seconds in all.
• With 1 cook or 8, the starting line is the same 0.5 seconds. A single set() wakes everyone who is waiting.

4-3. "Wake me when my ingredient arrives": Condition (10 min)

An Event is just on or off. But waiting in a kitchen is usually more specific than that. Minho is waiting for eggs, Sua for tofu. Every time something lands on the shelf, they have to check "is that mine?" and go back to waiting if it isn't. That's what Condition is for.

A Condition is a lock + a "wake me up" list. It guards the shared state (the ingredient shelf) with the lock, and when the state changes, it wakes the people on the list.

pantry = []                             # the ingredient shelf (shared state)
arrived = threading.Condition()         # a lock guarding the shelf + a "wake me up" list
start = time.perf_counter()

def cook(name, need):
    with arrived:
        ok = arrived.wait_for(lambda: need in pantry, timeout=0.5)
        print(f"{time.perf_counter() - start:.1f}s {name} {need if ok else 'gave up'}")

def supplier():
    for item in ["tofu", "egg"]:        # no fish today
        time.sleep(0.1)
        with arrived:
            pantry.append(item)
            arrived.notify_all()        # wake everyone who's waiting

threads = [threading.Thread(target=cook, args=a) for a in [("minho", "egg"), ("sua", "tofu"), ("tak", "fish")]]
threads.append(threading.Thread(target=supplier))
for th in threads:
    th.start()
for th in threads:
    th.join()
# 0.1s sua tofu
# 0.2s minho egg
# 0.5s tak gave up

• You look at and change the shelf only while holding the lock with with arrived:. Same rule as Lesson 3.
• wait_for(condition) waits until the condition is true. While waiting, it puts the lock down for a moment, so the supplier can put things on the shelf. On waking, it takes the lock again and checks the condition again.
• With a timeout, when time runs out it returns the condition's last value. The fish never came, so after 0.5 seconds Mr. Tak got False and gave up.
• notify_all() wakes everyone on the list. Each one checks their own condition and goes back to sleep if it isn't met. When the tofu arrived, Minho and Mr. Tak woke up too, but it wasn't their ingredient, so they went back to waiting.

notify() wakes just one waiter. It suits work where it doesn't matter who takes it, like "any server can take a finished plate."

plates = 0
counter = threading.Condition()

def cook():
    global plates
    time.sleep(0.1)
    with counter:
        plates += 1
        counter.notify()                # wake just one waiter

def waiter(name):
    global plates
    with counter:
        counter.wait_for(lambda: plates > 0)
        plates -= 1
        print(name, "took a plate")

ws = [threading.Thread(target=waiter, args=(n,)) for n in ["w1", "w2"]]
cs = [threading.Thread(target=cook) for _ in range(2)]
for th in ws + cs:
    th.start()
for th in ws + cs:
    th.join()
print(plates)
# e.g.:
# w1 took a plate
# w2 took a plate
# 0

Waking one person per plate means fewer cases of two people waking at once and fighting over a single plate. On the other hand, if people are waiting for different conditions, notify() may wake the wrong one, so use notify_all(). When in doubt, notify_all() is the safe choice.

4-4. Four plates for the group table, all at once: Barrier (5 min)

A four-top in the group section needs all four plates at the same time. If one goes out early, the other three guests sit there waiting while it gets cold. A Barrier(n) waits at the pass until all n people have arrived, then sends them out together.

start = time.perf_counter()
table = threading.Barrier(4, action=lambda: print("serve 4 plates"))
done = []

def cook(name, sec):
    time.sleep(sec)                     # everyone works at a different speed
    i = table.wait()                    # wait at the pass until all four arrive
    done.append((round(time.perf_counter() - start, 1), i, name))

crew = [("minho", 0.1), ("sua", 0.2), ("chef", 0.3), ("tak", 0.4)]
threads = [threading.Thread(target=cook, args=c) for c in crew]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(sorted(done))
# serve 4 plates
# [(0.4, 0, 'minho'), (0.4, 1, 'sua'), (0.4, 2, 'chef'), (0.4, 3, 'tak')]

• Minho finished at 0.1 seconds, but all four go through together at 0.4 seconds, when Mr. Tak, the slowest, arrives.
• wait() returns a different number to each person, from 0 to n-1. In CPython it's 0, 1, 2, 3 in order of arrival. You can use that number to pick "a job for just one person" (e.g., if i == 0: write the log).
• The function given as action= runs only once, when the last person arrives. It's the signal that the plates go out.
• Once everyone is through, the Barrier resets and can be used again for the next table. If wait(timeout=) runs out of time, the Barrier breaks and everyone waiting gets a BrokenBarrierError.

4-5. Tonight's tools at a glance (4 min)

All five live in threading, and all are built on top of the lock from Lesson 3. In fact Semaphore, Event and Barrier use a Condition internally, and a Condition holds a lock inside (an RLock unless you give it one).

Closing time (10 min)

Tonight's goal: nobody lights a burner before the open signal, no more than three burners at once, and six orders out in under a second. An Event sets the starting line, a Semaphore(3) guards the burners, and a Lock counts "how many are at the flames right now."

opened = threading.Event()
stoves = threading.Semaphore(3)
lock = threading.Lock()
on_fire = peak = 0
log = []
start = time.perf_counter()

def cook(n):
    global on_fire, peak
    opened.wait()                       # 1. no flames before "Open!"
    with stoves:                        # 2. at most three burners at once
        with lock:
            on_fire += 1
            peak = max(peak, on_fire)
        time.sleep(0.3)
        with lock:
            on_fire -= 1
            log.append((round(time.perf_counter() - start, 1), f"#{n}"))

threads = [threading.Thread(target=cook, args=(n,)) for n in range(1, 7)]
for th in threads:
    th.start()
time.sleep(0.3)                         # getting ready for guests
opened.set()
for th in threads:
    th.join()
print(sorted(log))
print(f"peak {peak}, total {time.perf_counter() - start:.2f}s")
# e.g.:
# [(0.6, '#1'), (0.6, '#2'), (0.6, '#3'), (0.9, '#4'), (0.9, '#5'), (0.9, '#6')]
# peak 3, total 0.90s

• Nobody cooks until 0.3 seconds. The first plates come out at 0.6 seconds and the last at 0.9, inside the one-second goal.
• peak is 3 every time you run it. Which three orders come out at 0.6 seconds may vary from run to run.
• on_fire and peak are values all six cooks change, so they're wrapped in lock, even inside the semaphore.

Near closing, a woman sitting alone at the window finishes a bowl of noodles. She checks her wristwatch once, writes something in a small notebook, pays quietly and leaves.

The chef shuts the burner valves one by one. "Everyone queued up nicely tonight. But what if, besides the burners, there's only one knife and one cutting board? If Minho holds the knife and waits for the board while Sua holds the board and waits for the knife, what happens?"

Exercises

1. On the burner timeline, set 6 cooks and 6 orders, change the burners from 1 to 6, and write the total times in a table. Why do 4 and 5 burners take the same time?
2. Build the "closing time" pattern from 4-2 yourself. Create closing = threading.Event(), and have a cook thread make one bowl every 0.1 seconds, stopping once closing.is_set(). The main thread calls closing.set() after 0.35 seconds and then join()s. How many bowls came out?
3. Think about what happens if you change Barrier(4) to Barrier(5) in the 4-4 Barrier example. So it doesn't really freeze forever, change it to table.wait(timeout=1), run it, and see which exception you get.