今夜の注文書
#04 · 四日目の夜
ピンチ
団体客のために料理人が6人に増えた。ところがコンロは3つだけ。しかもシェフが「オープン!」と叫ぶまでは、誰も火をつけてはいけない。
目標
オープンの合図の前には誰も火をつけず、コンロは同時に3つまでしか使わずに、6つの注文を1秒以内に出す。
深夜食堂 ドンシ
四日目の夜、9時40分。団体の予約が入っていて、一日だけの料理人が3人加わりました。料理人は6人、コンロは3つです。
ミンホが真っ先にコンロの前に立って火をつけます。「先に温めておけば速いでしょ」シェフは黙ってガスの元栓を閉めます。「オープン前に火をつけるのは禁止。それに、コンロの前に6人が群がってたら誰も動けないよ」
スアが手を挙げます。「コンロの前に番号札を3枚だけ掛けておくのはどうですか? 番号札を持っている人だけが火の前に立つんです」シェフが初めて笑います。「それが今夜の最初の道具だね」
3限目の Lock はチョークひとつ、つまり一度に一人でした。Semaphore(n) は番号札n枚です。同時にn人まで入ることができ、n+1人目は誰かが番号札を返すまで待ちます。
import threading
import time
stoves = threading.Semaphore(3) # コンロは3つ
start = time.perf_counter()
def cook(name):
with stoves: # 空いたコンロが出るまで待ってから取る
print(f"{time.perf_counter() - start:.1f}s {name} on")
time.sleep(0.3)
names = [f"cook{i}" for i in range(1, 7)]
threads = [threading.Thread(target=cook, args=(n,)) for n in names]
for th in threads:
th.start()
for th in threads:
th.join()
print(f"total {time.perf_counter() - start:.2f}s")
# 例:
# 0.0s cook1 on
# 0.0s cook2 on
# 0.0s cook3 on
# 0.3s cook4 on
# 0.3s cook5 on
# 0.3s cook6 on
# total 0.60swith stoves: に入るときに番号札を1枚持っていき(acquire)、出るときに返します(release)。番号札が0枚なら、入ろうとする側が待ちます。Semaphore(1) は Lock と同じく一度に一人です。ただし Lock は解放済みのロックをもう一度放すと RuntimeError になるのに対し、セマフォは release() を余分に呼んでも黙って番号札が増えます(下のBoundedSemaphore)。下のタイムラインでコンロの数を変えてみてください。黄色のバーはコンロを待っている時間、オレンジのバーは実際に調理している時間です。
厨房タイムライン
注文
6
料理人
3
コンロ
2
調理
待ち
キャンセル
失敗
Semaphore は何枚返してもらったかを数えません。バグで release() を一回多く呼ぶと、番号札がこっそり4枚になります。コンロは3つなのに、4人が火の前に立つことになります。
loose = threading.Semaphore(3)
loose.release() # うっかり一回多く返した
print(sum(loose.acquire(blocking=False) for _ in range(10))) # コンロが4つになった
# 4
strict = threading.BoundedSemaphore(3)
strict.release()
# ValueError: Semaphore released too many timesBoundedSemaphore は、最初に決めた数より多く返されると ValueError を出します。数を制限するためにセマフォを使うなら、ほとんどいつでも BoundedSemaphore のほうが向いています。ミスが黙って見過ごされず、その場で爆発してくれるからです。
シェフのひとこと
セマフォは数を数えてくれるだけで、黒板を守ってくれるわけじゃない。コンロ3つに3人立ってても、その3人が同じ黒板に数を書いたらまた競合状態。3人まで入れる部屋の中で共有の値に触るなら、その中でもう一度Lockを使って。
Event は厨房の壁にある信号灯ひとつです。最初は消えていて、誰かがつけると(set())、その合図を待っていた(wait())人が全員いっせいに動き出します。
opened = threading.Event() # 最初は消えている
start = time.perf_counter()
def cook(name):
opened.wait() # 合図がつくまで待つ
print(f"{time.perf_counter() - start:.1f}s {name} fire")
threads = [threading.Thread(target=cook, args=(n,)) for n in ["minho", "sua", "tak"]]
for th in threads:
th.start()
time.sleep(0.5)
print(opened.is_set())
opened.set() # 「オープン!」
for th in threads:
th.join()
print(opened.is_set())
# False
# 0.5s minho fire (3行の順番は実行ごとに変わる)
# 0.5s sua fire
# 0.5s tak fire
# Truewait() に時間の上限を渡すこともできます。このとき返ってくる値が大事です。合図がついていれば True、時間切れになってもまだ消えていれば False です。
closing = threading.Event()
print(closing.wait(timeout=0.2)) # 0.2秒待ったが合図がない
closing.set()
print(closing.wait(timeout=0.2)) # すでについていればすぐTrue
closing.clear() # また消す
print(closing.is_set())
# False
# True
# False| メソッド | すること |
|---|---|
set() | 合図をつける。待っていたスレッドが全員起きる |
wait() | 合図がつくまで待つ。すでについていればそのまま通過する |
wait(timeout) | その時間までしか待たない。ついていれば True、でなければ False |
clear() | 合図をまた消す。以降の wait() はまた待つ |
is_set() | いまついているかどうか |
よく使う形がひとつあります。「営業終了」の合図を Event にして、働くスレッドがループを回るたびに if closing.is_set(): break で確かめるようにするものです。2限目のdaemonのように途中でぶつ切りにせず、自分で片づけて終わるようにする方法です。
下のタイムラインは、シェフが0.5秒後に「オープン!」と叫ぶ厨房です。
厨房タイムライン
料理人
3
調理
待ち
キャンセル
失敗
set() 一回が、待っている人を全員起こすからです。Event はついているか消えているかの2つだけです。でも厨房の「待つ」は、たいていもっと具体的です。ミンホは卵を、スアは豆腐を待っています。棚に材料が入るたびに「自分の材料は来た?」と確かめ、まだなら待ち直さなければなりません。こういうときに Condition を使います。
Condition はロックひとつ + 「起こして」名簿です。共有の状態(材料の棚)をロックで守りながら、状態が変わったら名簿の人を起こします。
pantry = [] # 材料の棚(共有の状態)
arrived = threading.Condition() # 棚を守るロック + 「起こして」名簿
start = time.perf_counter()
def cook(name, need):
with arrived:
ok = arrived.wait_for(lambda: need in pantry, timeout=0.5)
print(f"{time.perf_counter() - start:.1f}s {name} {need if ok else 'gave up'}")
def supplier():
for item in ["tofu", "egg"]: # 魚は今日は来ない
time.sleep(0.1)
with arrived:
pantry.append(item)
arrived.notify_all() # 待っている人を全員起こす
threads = [threading.Thread(target=cook, args=a) for a in [("minho", "egg"), ("sua", "tofu"), ("tak", "fish")]]
threads.append(threading.Thread(target=supplier))
for th in threads:
th.start()
for th in threads:
th.join()
# 0.1s sua tofu
# 0.2s minho egg
# 0.5s tak gave upwith arrived: でロックを取っているあいだだけ棚を見たり変えたりします。3限目のルールそのままです。wait_for(条件) は条件が真になるまで待ちます。待っているあいだはロックをいったん手放すので、納品業者が棚に材料を入れられます。起きたらロックを取り直し、条件をもう一度確かめます。timeout を渡すと、時間切れのときに条件の最後の値を返します。魚は最後まで来なかったので、タクさんは0.5秒後に False を受け取ってあきらめました。notify_all() は名簿の全員を起こします。起きた人はそれぞれ自分の条件を確かめ、違えばまた眠ります。豆腐が届いたときミンホとタクさんも起きましたが、自分の材料ではないので待ち直しました。notify() は待っている人を一人だけ起こします。誰が持っていってもかまわない仕事、たとえば「できあがった皿を手の空いたホール係が1枚持っていく」に向いています。
plates = 0
counter = threading.Condition()
def cook():
global plates
time.sleep(0.1)
with counter:
plates += 1
counter.notify() # 待っている人を一人だけ起こす
def waiter(name):
global plates
with counter:
counter.wait_for(lambda: plates > 0)
plates -= 1
print(name, "took a plate")
ws = [threading.Thread(target=waiter, args=(n,)) for n in ["w1", "w2"]]
cs = [threading.Thread(target=cook) for _ in range(2)]
for th in ws + cs:
th.start()
for th in ws + cs:
th.join()
print(plates)
# 例:
# w1 took a plate
# w2 took a plate
# 0皿1枚につき一人しか起こさないので、二人が同時に起きて1枚の皿を取り合うことが減ります。逆に、人によって待っている条件が違うなら、notify() が見当違いの人を起こしてしまうことがあるので notify_all() を使います。迷ったら notify_all() が安全です。
シェフのひとこと
Conditionで wait() を使うときは、必ず while の中で条件を確かめ直して。起こされたからって、自分の材料が届いた保証はない。ほかの人の材料かもしれないし、誰かが先に持っていったかもしれない。wait_for(条件) を使えばその while を代わりに回してくれるから、それがいちばん楽だよ。
団体席の4人掛けテーブルには、4皿を同時に出さなければなりません。1皿だけ先に出すと、残りの3人は料理が冷めるのを待ちながら座っていることになりますから。Barrier(n) は、n人全員がそろうまでパスの前で待ち、そろったらいっせいに送り出します。
start = time.perf_counter()
table = threading.Barrier(4, action=lambda: print("serve 4 plates"))
done = []
def cook(name, sec):
time.sleep(sec) # それぞれ速さが違う
i = table.wait() # 4人がそろうまでパスの前で待つ
done.append((round(time.perf_counter() - start, 1), i, name))
crew = [("minho", 0.1), ("sua", 0.2), ("chef", 0.3), ("tak", 0.4)]
threads = [threading.Thread(target=cook, args=c) for c in crew]
for th in threads:
th.start()
for th in threads:
th.join()
print(sorted(done))
# serve 4 plates
# [(0.4, 0, 'minho'), (0.4, 1, 'sua'), (0.4, 2, 'chef'), (0.4, 3, 'tak')]wait() は0からn-1まで、人ごとに違う番号を返します。CPythonでは到着した順に0、1、2、3です。この番号で「一人だけがやる仕事」(例:if i == 0: でログを残す)を決められます。action= に渡した関数は、最後の一人が到着したときに一度だけ実行されます。皿を送り出す合図です。wait(timeout=) が時間切れになるとBarrierが壊れ、待っていた全員が BrokenBarrierError を受け取ります。| 道具 | 厨房でいえば | いつ使う |
|---|---|---|
Lock | チョークひとつ | 共有の値を一度に一人だけが書き直すとき(3限目) |
Semaphore(n)、BoundedSemaphore(n) | コンロ前の番号札n枚 | 同時にn個までしか使わせないとき |
Event | 壁の信号灯 | 「スタート」「終わり」のような一度きりの合図を全員に |
Condition | ロック + 「起こして」名簿 | 「この条件になったら起こして」。共有の状態が変わるのを待つとき |
Barrier(n) | 団体席のパス | n人がそろったら一緒に出発 |
5つとも threading にあり、どれも3限目のロックの上に建てた道具です。実際、Semaphore、Event、Barrier は内部で Condition を使っていて、Condition は中にロック(指定しなければ RLock)をひとつ持っています。
今夜の目標は、オープンの合図の前には誰も火をつけず、コンロは同時に3つまでしか使わずに、6つの注文を1秒以内に出すことでした。Event でスタートラインを、Semaphore(3) でコンロを、Lock で「いま火の前に何人いるか」を数えます。
opened = threading.Event()
stoves = threading.Semaphore(3)
lock = threading.Lock()
on_fire = peak = 0
log = []
start = time.perf_counter()
def cook(n):
global on_fire, peak
opened.wait() # 1. 「オープン!」の前には火をつけない
with stoves: # 2. コンロは同時に3つまで
with lock:
on_fire += 1
peak = max(peak, on_fire)
time.sleep(0.3)
with lock:
on_fire -= 1
log.append((round(time.perf_counter() - start, 1), f"#{n}"))
threads = [threading.Thread(target=cook, args=(n,)) for n in range(1, 7)]
for th in threads:
th.start()
time.sleep(0.3) # 客を迎える準備
opened.set()
for th in threads:
th.join()
print(sorted(log))
print(f"peak {peak}, total {time.perf_counter() - start:.2f}s")
# 例:
# [(0.6, '#1'), (0.6, '#2'), (0.6, '#3'), (0.9, '#4'), (0.9, '#5'), (0.9, '#6')]
# peak 3, total 0.90speak は何度回しても3です。0.6秒に出る3つの注文がどの番号かは、実行ごとに変わることがあります。on_fire と peak は6人の料理人が一緒に書き直す値なので、lock で包みました。セマフォの中であってもそうです。閉店間際、窓際の席に一人で座った女性客が、麺を一杯きれいに平らげます。腕時計にちらりと目をやり、小さな手帳に何かを書きつけると、静かに会計を済ませて出ていきました。
シェフがコンロの元栓をひとつずつ閉めながら言います。「今日はみんな、ちゃんと並べたね。でもコンロじゃなくて、包丁とまな板もひとつずつしかなかったら? ミンホが包丁を持ってまな板を待ってて、スアがまな板を持って包丁を待ってたら、どうなると思う?」
営業日誌 — 四日目の夜。料理人6人、コンロ3つ。オープンの合図(Event)の前は待機、コンロは番号札3枚(Semaphore)で制限し、0.9秒で6件完了。窓際の席にお客さんが一人、麺を一杯。
closing = threading.Event() を用意し、料理人スレッドが0.1秒ごとに1皿ずつ作りながら、closing.is_set() なら止まるようにします。メインスレッドは0.35秒後に closing.set() を呼んで join() します。何皿できたでしょう?Barrier(4) を Barrier(5) に変えたら何が起きるか考えてみましょう。本当に永遠に止まってしまわないように table.wait(timeout=1) に変えてから回し、どんな例外が出るか確かめてください。シェフのレシピカード
Semaphore(n):with sem: で同時にn個まで。数の制限なら BoundedSemaphore(n) で release のミスを ValueError としてつかまえる。
Event:set() でつけると wait() していた全員が起きる。wait(timeout) はついていれば True、でなければ False。clear()、is_set()。
Condition:with cond: の中で cond.wait_for(条件)。状態を変えた側が notify()(一人)か notify_all()(全員)。
Barrier(n):wait() でn人がそろったら一緒に通過。返す番号は0〜n-1、action= は一度だけ。
セマフォの中でも共有の値を書き直すならLock。道具は重ねて使う。
新しい授業・講座のお知らせを受け取る
この講座に授業が追加されたり、続きの講座が公開されたりしたらメールでお知らせします。
R
ReflexKR
触って学ぶPythonとReflex。すべての例はReflex 0.10.0で実際に動かして確認しています。
ReflexKRはバナー広告なし・無料で運営しています。役に立ったら、コーヒー一杯で応援してください。 ☕ コーヒーで応援する
アフィリエイトと支援について© 2026 ReflexKR · Since 2026 · CC BY-NC 4.0
質問と回答
この回について気になることを書き込んでください。すべての言語のページの質問がここに集まります。