出張料理人：concurrent.futures

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:   # 出張料理人3人
    fut = agency.submit(cook, 1)        # 仕事を任せると「引換券」(Future)をくれる
    print(fut.done())                   # まだ調理中
    print(fut.result())                 # 終わるまで待ってから結果を受け取る
    print(fut.done())
# False
# #1 ready
# True

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件
# withブロックを出たときには、すべての仕事が終わっている
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

from concurrent.futures import as_completed

def cook(n):
    time.sleep(0.1 * (5 - n))     # #1がいちばん時間がかかる (0.4秒 … 0.1秒)
    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):                 # 終わった順
        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)):      # 入れた順
        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

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

def cook(n):
    if n == 3:
        raise ValueError(f"#{n} burnt")    # 3番の注文を焦がした
    return f"#{n}"

with ThreadPoolExecutor(max_workers=2) as agency:
    futures = [agency.submit(cook, n) for n in range(1, 5)]
# ここまでは何事もなかったかのように静か

for f in futures:
    try:
        print(f.result())
    except ValueError as e:                # result()を呼んだときに送出される
        print("error:", e)
# #1
# #2
# error: #3 burnt
# #4

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

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

def ring_bell(fut):                      # 終わったら呼ばれる関数（コールバック）。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)
# 例：
# bell: #1 agency_0
# bell: #2 agency_1

with ThreadPoolExecutor(max_workers=1) as agency:     # 料理人1人
    futures = [agency.submit(time.sleep, 0.2) for _ in range(3)]
    time.sleep(0.05)                                  # 最初の注文が火にかかるすき
    print([f.cancel() for f in futures])              # もう始まったものは取り消せない
    print([f.cancelled() for f in futures], [f.done() for f in futures])
# [False, True, True]
# [False, True, True] [False, True, True]

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)     # 並んでいる注文は取り消し、火の上の注文は最後まで
print(sum(f.cancelled() for f in futures), sum(not f.cancelled() for f in futures))
# 4 2

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)   # ひとつでも終わったら
    print([f.result() for f in done], len(pending))
    done, pending = wait(futures, timeout=0.15)                  # デフォルトはALL_COMPLETED
    print(len(done), len(pending))
# [0.1] 2
# 2 1

from concurrent.futures import InterpreterPoolExecutor, ProcessPoolExecutor, ThreadPoolExecutor

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

if __name__ == "__main__":               # プロセスプールには必須（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]

def cook(n):
    time.sleep(0.1 + 0.05 * (n % 3))     # 0.1〜0.2秒
    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):     # できたものから出す
        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  (時間はおおよそ)