Agency cooks: concurrent.futures

Night seven, half past ten. The phone rings, and the chef sets the receiver down. "Company dinner. Thirty covers."

Minho reties his apron. "Should I make 30 threads?" The chef shakes her head. "Hire people and it's all on us: housing them, paying them, letting them go. Tonight we call a staffing agency. They manage the people. We just hand over the work."

7-1. Handing work to an agency: ThreadPoolExecutor (10 min)

concurrent.futures.ThreadPoolExecutor is a staffing agency that keeps a few threads on hand and gives each incoming job to whichever thread is free. It takes over everything you did by hand in Lesson 2: creating each Thread, start, join, and setting up a list for the results.

import time
from concurrent.futures import ThreadPoolExecutor

def cook(n):
    time.sleep(0.2)
    return f"#{n} ready"

with ThreadPoolExecutor(max_workers=3) as agency:   # three agency cooks
    fut = agency.submit(cook, 1)        # hand over a job and get a claim ticket (Future)
    print(fut.done())                   # still cooking
    print(fut.result())                 # wait until it's done, then take the result
    print(fut.done())
# False
# #1 ready
# True

• submit(function, args...) hands over the job and immediately returns a Future. It isn't the food; it's a "claim ticket for the food." Print it with print(fut) and you'll see its state, something like <Future at 0x... state=running>.
• fut.result() waits until the dish is done and returns the function's return value. This solves the Lesson 2 problem where "threads don't give back a return value" and we had to stash results in a list.
• When you leave the with block, shutdown(wait=True) is called: it waits for every job you handed over to finish and then cleans up the threads.

Now let's hand over 30. There are only five cooks, so each one takes six orders in turn.

import threading

def cook(n):
    time.sleep(0.1)
    return threading.current_thread().name

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5, thread_name_prefix="agency") as agency:
    futures = [agency.submit(cook, n) for n in range(1, 31)]   # 30 orders
# once you're out of the with block, every job is done
names = {f.result() for f in futures}
print(len(futures), sorted(names), f"{time.perf_counter() - start:.2f}s")
# 30 ['agency_0', 'agency_1', 'agency_2', 'agency_3', 'agency_4'] 0.60s

Thirty orders, but only five threads were ever created. Thirty 0.1-second orders split among five people: 0.6 seconds. You pay the cost of creating and destroying threads only once, and the same cooks keep taking the next order. This is called a thread pool.

7-2. Finish order vs. submit order: as_completed and map (10 min)

There are two ways to collect results. This time, let's make order #1 take the longest.

from concurrent.futures import as_completed

def cook(n):
    time.sleep(0.1 * (5 - n))     # #1 takes the longest (0.4 s … 0.1 s)
    return f"#{n}"

with ThreadPoolExecutor(max_workers=4) as agency:
    start = time.perf_counter()
    futures = [agency.submit(cook, n) for n in range(1, 5)]
    for f in as_completed(futures):                 # finish order
        print(f.result(), f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
    start = time.perf_counter()
    for dish in agency.map(cook, range(1, 5)):      # submit order
        print(dish, f"{time.perf_counter() - start:.1f}s", end="  ")
    print()
# #4 0.1s  #3 0.2s  #2 0.3s  #1 0.4s
# #1 0.4s  #2 0.4s  #3 0.4s  #4 0.4s

Even though #4 finished at 0.1 seconds, map holds it back until #1 finishes at 0.4 seconds. It has to keep the submit order. The total time is the same, but when the first plate goes out is different.

as_completed hands back only Futures, so to know which order a result belongs to, the usual trick is to pair them up in a dict.

with ThreadPoolExecutor(max_workers=4) as agency:
    table_of = {agency.submit(cook, n): f"T{n}" for n in range(1, 5)}   # Future → table
    for fut in as_completed(table_of):
        print(table_of[fut], fut.result())
# T4 #4
# T3 #3
# T2 #2
# T1 #1

The "Agency cooks" mode in the timeline below also runs on a real ThreadPoolExecutor. In this scenario, the later the order number, the shorter the order takes.

• Both modes take about 1.0 seconds, and the bars look almost the same. The cooks work exactly the same way; only the order you receive results in changes.
• With as_completed, the returned order is jumbled, for example #3 → #2 → #1 → #6 → #5 → #4 → …. Of the orders the three cooks start together, the short ones finish first.
• With map, it's always #1 → #2 → … → #10.
• Raise the cooks to the maximum of 8 and it's done in about 0.4 seconds, with as_completed returning them nearly backwards, like #8 → #7 → … → #1. Eight orders start at once, and the short, later-numbered ones finish first.

7-3. A burnt plate shows up when you redeem the ticket (7 min)

If an exception happens inside a job you handed over, it is quietly stored in the Future. The moment you call result(), it blows up again right there.

def cook(n):
    if n == 3:
        raise ValueError(f"#{n} burnt")    # order #3 got burnt
    return f"#{n}"

with ThreadPoolExecutor(max_workers=2) as agency:
    futures = [agency.submit(cook, n) for n in range(1, 5)]
# so far, all quiet, as if nothing happened

for f in futures:
    try:
        print(f.result())
    except ValueError as e:                # it blows up when you call result()
        print("error:", e)
# #1
# #2
# error: #3 burnt
# #4

print(repr(futures[2].exception()), futures[0].exception())
# ValueError('#3 burnt') None

• If you never call result(), the exception isn't printed anywhere and just vanishes. A burnt plate never reaches the guest, and nobody even knows.
• exception() hands you the exception without raising it. It's None if the job succeeded.
• map raises the exception right where you hit it while pulling results, and you can't get any results after that. If you need the rest even when one fails, use submit with as_completed.

To do something the moment a job finishes, hang a bell on it with add_done_callback. You pass a function that takes the finished Future as its one argument.

def cook(n):
    time.sleep(0.1)
    return f"#{n}"

def ring_bell(fut):                      # called when done (a callback); takes one Future
    print("bell:", fut.result(), threading.current_thread().name)

with ThreadPoolExecutor(max_workers=2, thread_name_prefix="agency") as agency:
    for n in (1, 2):
        agency.submit(cook, n).add_done_callback(ring_bell)
# e.g.
# bell: #1 agency_0
# bell: #2 agency_1

The callback runs in the thread of the agency cook who finished the job. (Attach it to a Future that's already done and it runs right away, in the thread that attached it.) So if the callback changes shared data, the locking rules from Lesson 3 apply just the same. An exception inside a callback is only logged and otherwise ignored.

7-4. Cancelling, and waiting for just one (8 min)

cancel() can cancel only jobs that haven't started yet. A dish already on the heat can't be taken back.

with ThreadPoolExecutor(max_workers=1) as agency:     # one cook
    futures = [agency.submit(time.sleep, 0.2) for _ in range(3)]
    time.sleep(0.05)                                  # time for the first order to hit the heat
    print([f.cancel() for f in futures])              # what's already started can't be taken back
    print([f.cancelled() for f in futures], [f.done() for f in futures])
# [False, True, True]
# [False, True, True] [False, True, True]

• cancel() returns True if the cancellation worked, and False if the job is already running or finished.
• done() is True once a job has finished or been cancelled. The first order is still cooking, so it's False.

To end the contract with the agency all at once, use shutdown(cancel_futures=True). Every job still in line is cancelled, and it waits only for the ones already started.

agency = ThreadPoolExecutor(max_workers=2)
futures = [agency.submit(time.sleep, 0.2) for _ in range(6)]
time.sleep(0.05)
agency.shutdown(cancel_futures=True)     # orders in line are cancelled, orders on the heat finish
print(sum(f.cancelled() for f in futures), sum(not f.cancelled() for f in futures))
# 4 2

When you want to move as soon as any one of several jobs is done, use wait.

from concurrent.futures import FIRST_COMPLETED, wait

def cook(sec):
    time.sleep(sec)
    return sec

with ThreadPoolExecutor(max_workers=3) as agency:
    futures = [agency.submit(cook, s) for s in (0.3, 0.1, 0.2)]
    done, pending = wait(futures, return_when=FIRST_COMPLETED)   # as soon as any one is done
    print([f.result() for f in done], len(pending))
    done, pending = wait(futures, timeout=0.15)                  # the default is ALL_COMPLETED
    print(len(done), len(pending))
# [0.1] 2
# 2 1

wait returns two sets: the finished and the unfinished. return_when can be FIRST_COMPLETED, FIRST_EXCEPTION or ALL_COMPLETED (the default), and once timeout passes it returns even if not everything is done. It doesn't raise anything, so check for yourself whether pending is empty.

7-5. Same contract, different agency (5 min)

The nice thing about concurrent.futures is that the contract is the same. Keep submit, map, Future and as_completed as they are, and just swap the agency's name.

from concurrent.futures import InterpreterPoolExecutor, ProcessPoolExecutor, ThreadPoolExecutor

def knead(n):
    return sum(range(n))

if __name__ == "__main__":               # required for a process pool (Lesson 8)
    for Agency in (ThreadPoolExecutor, ProcessPoolExecutor, InterpreterPoolExecutor):
        with Agency(max_workers=2) as agency:
            print(Agency.__name__, list(agency.map(knead, [10, 100, 1000])))
# ThreadPoolExecutor [45, 4950, 499500]
# ProcessPoolExecutor [45, 4950, 499500]
# InterpreterPoolExecutor [45, 4950, 499500]

Save this example as a .py file and run it, not in the interactive shell. We'll see tomorrow night why a process pool insists on if __name__ == "__main__":.

If you leave out max_workers, a thread pool calls in min(32, os.process_cpu_count() + 4) workers (os.cpu_count() up to 3.12). On this 12-core PC, that's 16. Waiting work needs more threads than there are cores, hence the +4, and the cap at 32 keeps it from getting out of hand. In a real service, it's better to set it yourself based on how much the other server can take.

Closing time (10 min)

Tonight's goal was to hand 30 orders to six agency cooks, send each one out as soon as it's ready, all in under a second, without missing a burnt plate. Each order takes 0.1 to 0.2 seconds, and order 13 gets burnt.

def cook(n):
    time.sleep(0.1 + 0.05 * (n % 3))     # 0.1 to 0.2 s
    if n == 13:
        raise ValueError(f"#{n} burnt")
    return f"#{n}"

served, burnt = [], []
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=6, thread_name_prefix="agency") as agency:
    futures = {agency.submit(cook, n): n for n in range(1, 31)}
    for fut in as_completed(futures):     # send each out as soon as it's done
        try:
            served.append(fut.result())
        except ValueError:
            burnt.append(futures[fut])
print(len(served), burnt, f"{time.perf_counter() - start:.2f}s")
# 29 [13] 0.80s  (timing is approximate)

Twenty-nine plates went out as they finished, and the burnt number 13 went on the remake list. 0.8 seconds. No code to create threads, and no code to set up a results list in advance.

The guest by the window is here again tonight, alone, with a single bowl of noodles in front of her. While the group's plates stream out, she glances at her wristwatch once, then jots something brief in a small notebook.

As she sends the agency cooks home, the chef says, "Tomorrow we've got a mountain of dough. If I call in four agency cooks to knead it, will it go four times faster?"

Practice

1. In the closing-time code, replace as_completed with agency.map(cook, range(1, 31)). What happens at number 13, and how many plates can you collect?
2. In the timeline, drop the cooks to 1 (it takes about 2.8 seconds) and compare as_completed with map. Why does the returned order come out the same?
3. Using wait(..., return_when=FIRST_COMPLETED) and shutdown(cancel_futures=True), write code that takes only the first finished plate out of 30 orders and cancels the rest. How many dishes had already started?