ReflexKR

深夜厨房の並行処理

講座の概要

初日の行列：並行と並列

料理人を増やす：threadingの基本

黒板の数が合わない：競合状態とLock

コンロは3つだけ：同期の道具

包丁とまな板：デッドロック

伝票差し：queue

出張料理人：concurrent.futures

生地はなぜ速くならない：GILと並列

一人で回すホール係：asyncioの第一歩

団体予約のキャンセル：構造化された並行処理

止まったホール：ブロッキングと組み合わせ

評論家の夜：総まとめ

深夜厨房の並行処理

2限目

時間 0:50–1:40

料理人を増やす：threadingの基本

今夜の注文書

#02 · 二日目の夜

料理人を増やす

ピンチ

ミンホ、スア、タクさんが初出勤。3人ともよく動いているのに、誰がいつ何を仕上げたのか誰にもわからない。

目標

6つの注文を3人で分けて0.6秒以内に終わらせ、誰が何秒に何を仕上げたかを記録に残す。

深夜食堂ドンシ

二日目の夜、9時半。ラインコックの3人がエプロンを締めます。ミンホはもう包丁を握り、スアは注文書を順番に並べ、タクさんはゆっくりとコンロを拭いています。

10時になると注文が入り、3人はそれぞれ鍋を火にかけます。ところが30分後、シェフが「3番テーブルのうどん、誰が作った？」と聞くと、誰も答えられません。

シェフが首のタイマーを指でとんと叩きます。「料理人を雇うところまではできた。今日は名札をつけて、仕事を任せて、終わるまで待つやり方を覚えるよ」

2-1. 料理人を一人迎える：Thread、start、join (10分)

Pythonでは、料理人一人がスレッド（thread）ひとつです。同じ厨房（メモリ）を共有しながら、それぞれが自分の仕事を別々に進めます。スレッドは threading.Thread で作ります。

import threading
import time

def cook(name, dish):
    print(f"{name}: start {dish}")
    time.sleep(0.3)          # スープが煮えるのを待つ
    print(f"{name}: done {dish}")

minho = threading.Thread(target=cook, args=("minho", "ramen"))
minho.start()                # ミンホが仕事を始める
print("chef: next order")    # シェフは待たずに次の仕事へ
minho.join()                 # ミンホが終わるまで待つ
print("chef: served")
# minho: start ramen
# chef: next order
# minho: done ramen
# chef: served

target= にはスレッドがする仕事（関数）を、args= にはその関数に渡す引数をタプルで渡します。関数を呼ぶのではなく関数そのものを渡すので、cook() ではなく cook です。
Thread(...) を作っただけでは何も起きません。料理人を雇っただけの状態です。start() を呼んではじめて、新しいスレッドで cook が動き出します。
start() はすぐに戻ってきます。だからミンホがラーメンを作っているあいだに、シェフ（プログラムを最初に始めたメインスレッド）は次の行へ進みます。
join() はそのスレッドが終わるまで待ちます。「ミンホが皿を差し出すまで、パスの前で待つ」ということです。

3人にひとつずつ任せると、こうなります。

def cook(name, dish):
    time.sleep(0.3)
    print(f"{name}: done {dish}")

crew = [("minho", "ramen"), ("sua", "udon"), ("tak", "soba")]
threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.start()        # まず3人全員を送り出して
for th in threads:
    th.join()         # それから全員を待つ
print(f"{time.perf_counter() - start:.2f}s")
# 例：
# sua: done udon
# minho: done ramen
# tak: done soba
# 0.30s

3人が終わる順番は、実行するたびに変わります。3人とも0.3秒後にほぼ同時に終わるので、OSが誰を先に起こすか次第だからです。並行処理のコードでは「出力の順番」は約束ではありません。

start() と join() を同じループの中に書いてはいけません。th.start(); th.join() を交互にやると、一人が終わらないと次の人が出発できないので、結局0.9秒かかります。

待つ時間に上限をつけることもできます。join(timeout=秒) はその時間だけ待って戻り、まだ仕事中かどうかは is_alive() で確かめます。

slow = threading.Thread(target=time.sleep, args=(1.0,))
slow.start()
slow.join(timeout=0.2)
print(slow.is_alive())   # True：まだ煮えている途中
slow.join()
print(slow.is_alive())   # False

厨房タイムライン

注文

料理人

調理

待ち

キャンセル

失敗

注文6つ、それぞれ0.3秒。「料理人を増やす」で料理人を1人から6人まで変えてみてください。列がひとつずつ増え、全体の時間が1.8秒 → 0.9秒 → 0.6秒 → 0.3秒と縮みます。

料理人が1人なら「一人で順番に」とまったく同じで約1.8秒です。スレッドをひとつしか使わなければ、速くなる要素はありません。
料理人2人で約0.9秒、3人で約0.6秒、6人で約0.3秒です。
4人と5人はどちらも約0.6秒です。注文6つを4人で分けると、誰かが2つ担当しなければならないからです。かかる時間はいちばん多く受け持った人が決めます。
1列がスレッド1本です。このタイムラインの料理人たちも、上のコードと同じように Thread を作って start()、join() を呼んでいます。

2-2. run()を呼ぶとただの関数呼び出し (5分)

Thread には run() というメソッドもあります。名前だけ見ると「実行」ですが、直接呼ぶと落とし穴にはまります。

def cook(name, dish):
    time.sleep(0.3)
    print(f"{threading.current_thread().name}: done {dish}")

threads = [threading.Thread(target=cook, args=c) for c in crew]

start = time.perf_counter()
for th in threads:
    th.run()          # start()ではなくrun()
print(f"{time.perf_counter() - start:.2f}s")
# MainThread: done ramen
# MainThread: done udon
# MainThread: done soba
# 0.90s

run() は target を呼ぶだけの普通のメソッドです。新しいスレッドは作らず、今のスレッドで順番に実行します。だから3回とも MainThread がこなし、時間も 0.3 × 3 = 0.9秒です。
新しいスレッドを立ち上げるのは start() です。start() が新しいスレッドを作り、そのスレッドの中で run() を呼びます。

スレッドは一度しか出発できません。同じスレッドに start() を2回呼ぶとエラーになります。同じ仕事をもう一度したいなら、Thread を新しく作ってください。

th = threading.Thread(target=time.sleep, args=(0.1,))
th.start()
th.start()
# RuntimeError: threads can only be started once

シェフのひとこと

args=("ramen")にはカンマがないから、タプルじゃなくてただの文字列。すると r、a、m、e、n の5文字がばらばらに渡って、'cook() takes 1 positional argument but 5 were given' が出る。引数がひとつなら、必ず args=("ramen",) みたいにカンマを打って。それと、スレッドの中で起きた例外はシェフのところまで上がってこない。画面に 'Exception in thread ...' とトレースバックが出るだけで、プログラムはそのまま続いちゃう。見逃さないように、ログは必ず見ること。

2-3. 名札をつける：name、current_thread、enumerate (7分)

今夜の問題は「誰がやったかわからない」ことでした。スレッドには名札をつけられます。

def cook(dish):
    me = threading.current_thread()       # このコードを実行中のスレッド
    print(me.name, dish, threading.get_native_id())
    time.sleep(0.2)

a = threading.Thread(target=cook, args=("ramen",), name="minho")
b = threading.Thread(target=cook, args=("udon",))      # 名前を渡さないと？
a.start(); b.start()
print([t.name for t in threading.enumerate()])
print(threading.active_count())
a.join(); b.join()
print(threading.current_thread() is threading.main_thread())
print([t.name for t in threading.enumerate()])
# 例：
# minho ramen 25680
# Thread-1 (cook) udon 38072
# ['MainThread', 'minho', 'Thread-1 (cook)']
# 3
# True
# ['MainThread']

道具	返すもの
`Thread(name="minho")`	スレッドに名前をつける。渡さなければ `Thread-1 (cook)` のように番号と関数名がつく
`threading.current_thread()`	いまこの行を実行しているスレッドのオブジェクト
`threading.main_thread()`	プログラムを始めたメインスレッド（名前は `MainThread`）。今夜の厨房ではシェフ
`threading.enumerate()`	いま生きているスレッドの一覧（メインスレッドを含む）
`threading.active_count()`	その一覧の長さ
`threading.get_native_id()`	OSがつけたスレッド番号。Linuxの `top -H` などのシステムツールに出る番号と同じ

名前を渡さなかった b が Thread-2 ではなく Thread-1 なのは、番号が名前のないスレッドにだけ振られるからです。
get_native_id() の数字は実行するたびに変わります。Pythonの中で使う番号は th.ident（または threading.get_ident()）で、どちらもスレッドが出発したあとでないと決まりません。
仕事を終えたスレッドは enumerate() から消えます。退勤した料理人は名簿に載っていません。

2-4. スレッドは皿を返してくれない (6分)

関数は return で結果を返しますが、スレッドはその値を受け取ってくれません。target 関数の戻り値はそのまま捨てられ、join() も何も返しません。

def make(dish):
    time.sleep(0.1)
    return f"{dish} ok"

th = threading.Thread(target=make, args=("ramen",))
th.start()
print(th.join())     # joinは何も返さない
# None

そこで、結果を受け取る器をあらかじめ用意しておき、スレッドにそこへ盛ってもらいます。いちばん簡単なのは、料理人ごとに自分の枠を決めてあげる方法です。

dishes = ["ramen", "udon", "soba"]
results = [None] * len(dishes)          # 枠を先に用意する

def cook(i, dish):
    results[i] = make(dish)             # 自分の枠にだけ書く

threads = [threading.Thread(target=cook, args=(i, d)) for i, d in enumerate(dishes)]
for th in threads:
    th.start()
for th in threads:
    th.join()
print(results)
# ['ramen ok', 'udon ok', 'soba ok']

名前で引きたいなら dict を使います。

done = {}

def cook2(name, dish):
    done[name] = make(dish)

ts = [threading.Thread(target=cook2, args=(n, d)) for n, d in [("minho", "ramen"), ("sua", "udon")]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(sorted(done.items()))
# [('minho', 'ramen ok'), ('sua', 'udon ok')]

別々の枠、別々のキーに書くなら、料理人同士はぶつかりません。同じ枠の値を読んで書き直し始めると話が変わってきますが、それが明日の夜の事件です。
結果は必ず join() が終わってから読みます。その前に読むと、まだ None のままの枠があるかもしれません。
毎回器を用意するのが面倒なら、7限目の ThreadPoolExecutor が返す Future がこの仕事を代わりにやってくれます。future.result() で皿を直接受け取れます。

2-5. daemon=True：明かりを消すとき待ってくれないスレッド (6分)

Pythonのプログラムは、メインスレッドが終わっても、daemonでないスレッドが全部終わるまで待ってから終了します。普通に作ったスレッドはどれもdaemonではありません。

def clean():
    for i in range(1, 6):
        print("cleaning", i)
        time.sleep(0.1)

th = threading.Thread(target=clean)      # daemon=False（デフォルト）
th.start()
time.sleep(0.25)
print("lights off")
# cleaning 1
# cleaning 2
# cleaning 3
# lights off
# cleaning 4
# cleaning 5

シェフが「明かり消すよ」と声をかけても、掃除中の料理人が終わるまで店の扉は閉まりません。同じコードに daemon=True だけつけてみます。

th = threading.Thread(target=clean, daemon=True)
th.start()
time.sleep(0.25)
print("lights off", th.daemon)
# cleaning 1
# cleaning 2
# cleaning 3
# lights off True

daemonスレッドは、プログラムが終わるときに待ってもらえません。掃除の途中で明かりが消え、4番と5番は永遠にできずじまいです。
待ちたいなら、daemonでも join() を呼べばいいのです。join() はdaemonかどうかに関係なく待ちます。
daemonは「途中で切れてもかまわない仕事」にだけ使います。状態を定期的に表示する監視スレッドのようなものです。ファイルの書き込みや注文の記録のように最後までやり遂げるべき仕事をdaemonに任せると、書きかけのファイルが残ります。
daemonかどうかは start() の前に決めます。指定しなければ作ったスレッドの値を引き継ぎますが、メインスレッドはdaemonではないので、普通は False です。

2-6. そのほかの道具：Threadの継承、Timer、local (6分)

Threadを継承してrunを書き換える。料理人に自分の状態（担当の料理、結果）を持たせておきたいときに使います。2-2で見たとおり start() が新しいスレッドで run() を呼ぶので、run() だけ書き直せば十分です。

class Cook(threading.Thread):
    def __init__(self, name, dish):
        super().__init__(name=name)     # 必ず親の__init__から
        self.dish = dish
        self.result = None

    def run(self):                      # start()が新しいスレッドでこれを呼ぶ
        time.sleep(0.1)
        self.result = f"{self.dish} ok"

tak = Cook("tak", "soba")
tak.start()
tak.join()
print(tak.name, tak.result)
# tak soba ok

threading.Timer：n秒後に一度だけ。麺ゆでタイマーのように、決めた時間が過ぎたら関数を一度呼ぶスレッドです。鳴る前なら cancel() で止められます。

start = time.perf_counter()

def ring(msg):
    print(f"{time.perf_counter() - start:.1f}s", msg)

noodles = threading.Timer(0.5, ring, args=("noodles up",))
noodles.start()                     # 0.5秒後に一度鳴る
eggs = threading.Timer(1.0, ring, args=("eggs up",))
eggs.start()
eggs.cancel()                       # 鳴る前に止めれば何も起きない
noodles.join()
eggs.join()
print("end")
# 0.5s noodles up
# end

threading.local()：料理人ごとに別々のエプロンのポケット。同じ名前の変数なのに、スレッドごとに違う値が入ります。誰が入れたかによってポケットが違うのです。

pocket = threading.local()

def cook(name, n):
    pocket.name = name
    pocket.count = 0
    for _ in range(n):
        pocket.count += 1
        time.sleep(0.01)
    print(pocket.name, pocket.count)

ts = [threading.Thread(target=cook, args=a) for a in [("minho", 3), ("sua", 5)]]
for t in ts:
    t.start()
for t in ts:
    t.join()
print(hasattr(pocket, "name"))      # シェフ（メインスレッド）のポケットは空
# minho 3
# sua 5
# False

pocket はグローバル変数ひとつなのに、ミンホの count とスアの count は混ざりません。スレッドごとにデータベース接続をひとつずつ持たせるときによく使う方法です。

シェフのひとこと

スレッドを作るときは、まず名前をつけて。明け方にログを開いたら Thread-7、Thread-12 ばっかり並んでたら、誰がやらかしたのか絶対わからない。logging のフォーマットに %(threadName)s をひとつ入れておくだけで、行ごとに名前が出るから。

閉店後 (10分)

今夜の目標は、6つの注文を3人で分けて0.6秒以内に終わらせ、誰が何秒に何を仕上げたかを残すことでした。名札（name=）をつけ、仕上げるたびに共用の日誌に1行ずつ書いてもらいます。

start = time.perf_counter()
log = []                                # (仕上がった時刻, 料理人, 注文)

def cook(dishes):
    me = threading.current_thread().name
    for dish in dishes:
        time.sleep(0.3)
        log.append((round(time.perf_counter() - start, 1), me, dish))

plan = {"minho": ["#1", "#4"], "sua": ["#2", "#5"], "tak": ["#3", "#6"]}
threads = [threading.Thread(target=cook, args=(d,), name=n) for n, d in plan.items()]
for th in threads:
    th.start()
for th in threads:
    th.join()

for row in sorted(log):
    print(*row)
print(f"total {time.perf_counter() - start:.2f}s, {len(log)} orders")
# 0.3 minho #1
# 0.3 sua #2
# 0.3 tak #3
# 0.6 minho #4
# 0.6 sua #5
# 0.6 tak #6
# total 0.60s, 6 orders

引数がリストひとつなので、args=(d,) にカンマがあります。
log.append(...) は一度で終わるリスト操作なので、複数のスレッドが一緒に呼んでも行が消えることはありません。日誌は join() のあとで並べ替えて読みます。

シェフは日誌にざっと目を通してうなずきます。「よし、これで誰が何をしたかはわかる。じゃあ明日はもうひとつだけ試そう。4人がそれぞれ作った数を黒板のひとつの数字に足していったら、その数字は合ってると思う？」

営業日誌 — 二日目の夜。ミンホ、スア、タクさんが初出勤。名札をつけ、注文6件を2件ずつ分けて0.6秒で完了。誰が何秒に何を出したかを日誌に残した。

練習問題

2-1の3人の料理人の例で、start() と join() を同じループの中に一緒に書くように変え、何秒になるか予想してから測ってみましょう。
閉店後のコードで、タクさんだけ注文ひとつに0.5秒かかるようにしてみましょう（ヒント：料理人の名前によって sleep の時間を変える）。全体の時間を決めるのは誰でしょう？タイムラインで1列だけ長くなる様子と比べてみてください。
threading.Timer で「0.3秒ごとに1回、計3回」鳴る厨房タイマーを作ってみましょう。Timerは一度しか鳴らないので、鳴ったときに次のTimerを新しくスタートさせる必要があります。

シェフのレシピカード

th = threading.Thread(target=関数, args=(引数,), name="minho") → th.start() → th.join()
run()を直接呼ぶと、新しいスレッドなしに今のスレッドで順番に実行。スレッドはstart()で、一度だけ。
誰がやっているか：current_thread().name、main_thread()、enumerate()、get_native_id()
スレッドは戻り値を返さない。先に用意したリストの枠やdictのキーに入れ、join()のあとで読む。
daemon=Trueはプログラムの終了時に待ってもらえない。最後までやるべき仕事には使わない。
Threadの継承（runのオーバーライド）、Timer（n秒後に一度、cancel）、local()（スレッドごとに別々のポケット）

← 前へ

1限目 · 初日の行列：並行と並列

3限目 · 黒板の数が合わない：競合状態とLock

新しい授業・講座のお知らせを受け取る

この講座に授業が追加されたり、続きの講座が公開されたりしたらメールでお知らせします。