lesson 187 Threading & Multiprocessing · advanced

Executors and futures

concurrent.futures hides the threads: an executor runs submitted calls on a pool of workers and hands back futures, objects that will hold each result or exception. One interface covers thread pools and process pools.

Read and diagnose: this can't run in the browser, so you'll read real code and output and answer questions about it.

You'll be able to

Use submit and a Future's result(), exception() and done()
Choose between map (results in input order) and as_completed (results as they finish)
Write functions that accept any Executor, so they can be tested without threads

The idea

Starting and joining threads by hand gets repetitive, and unlimited threads overload servers. concurrent.futures.ThreadPoolExecutor(max_workers=8) keeps a pool of 8 threads; executor.submit(func, *args) queues a call and immediately returns a Future. future.result() waits for the call and returns its value, or re-raises the exception it raised. Using the executor in a with block waits for all work and shuts the pool down.

All executors share the Executor interface. This page has no threads, so the example defines a tiny real Executor subclass that runs each call immediately, in this thread: no concurrency, but the same futures, results and exceptions. On a computer you would pass a ThreadPoolExecutor to the same functions.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Futures hold results and exceptions

from concurrent.futures import Executor, Future

class InlineExecutor(Executor):
    """Runs each call as soon as it is submitted, in this thread. Same interface as ThreadPoolExecutor."""

    def submit(self, fn, *args, **kwargs):
        future = Future()
        try:
            future.set_result(fn(*args, **kwargs))
        except Exception as error:
            future.set_exception(error)
        return future

def stock_level(sku):
    levels = {"SPO-11681": 4, "HOM-22847": 0}
    return levels[sku]

with InlineExecutor() as executor:
    ok = executor.submit(stock_level, "SPO-11681")
    missing = executor.submit(stock_level, "BOO-58412")
    print(ok.done(), ok.result())
    print(type(missing.exception()).__name__)
    try:
        missing.result()
    except KeyError as error:
        print("result() re-raised", repr(error))
    print(list(executor.map(stock_level, ["HOM-22847", "SPO-11681"])))

example-2.py · A real thread pool in this page

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as pool:
    print(pool.submit(len, "SPO-11681").result())

Two ways to collect many results. executor.map(func, items) yields results in input order, waiting for each in turn, and raises the first exception it meets. concurrent.futures.as_completed(futures) yields futures as they finish, which is what you want to show progress or react to the first answer. On a real pool the finishing order depends on timing; never rely on it. For many jobs, keep a dictionary from future to its input so you know which job each finished future belongs to.

When to use it: almost always instead of raw threads: bounded workers, results and exceptions handled for you, and the same code works with ProcessPoolExecutor (“Processes: parallel work, separate memory”). Mistakes to avoid: never calling result() (exceptions are silently kept in the future); calling result() right after each submit (serial again); and assuming as_completed order.

Review

executor.submit(f, x) returns a Future; result() waits and returns the value or re-raises the error.
map keeps input order; as_completed yields futures as they finish.
Accept an Executor parameter: thread pool in production, an inline executor in tests.
Next: what goes wrong when threads share data, and locks.

Practise this