출장 요리사 파견: 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:   # 출장 요리사 세 명
    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:     # 요리사 한 명
    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  (시간은 대략)