R

ReflexKR

Reflex公式ドキュメント

深夜厨房の並行処理

·

講座の概要

1

初日の行列:並行と並列

2

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

3

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

4

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

5

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

6

伝票差し:queue

7

出張料理人:concurrent.futures

8

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

9

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

10

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

11

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

12

評論家の夜:総まとめ

深夜厨房の並行処理
4限目

時間 2:30–3:20

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

今夜の注文書

#04 · 四日目の夜

コンロは3つだけ

ピンチ

団体客のために料理人が6人に増えた。ところがコンロは3つだけ。しかもシェフが「オープン!」と叫ぶまでは、誰も火をつけてはいけない。

目標

オープンの合図の前には誰も火をつけず、コンロは同時に3つまでしか使わずに、6つの注文を1秒以内に出す。

深夜食堂 ドンシ

四日目の夜、9時40分。団体の予約が入っていて、一日だけの料理人が3人加わりました。料理人は6人、コンロは3つです。

ミンホが真っ先にコンロの前に立って火をつけます。「先に温めておけば速いでしょ」シェフは黙ってガスの元栓を閉めます。「オープン前に火をつけるのは禁止。それに、コンロの前に6人が群がってたら誰も動けないよ」

スアが手を挙げます。「コンロの前に番号札を3枚だけ掛けておくのはどうですか? 番号札を持っている人だけが火の前に立つんです」シェフが初めて笑います。「それが今夜の最初の道具だね」

4-1. コンロ3つ:Semaphore (12分)

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.60s
  • with stoves: に入るときに番号札を1枚持っていき(acquire)、出るときに返します(release)。番号札が0枚なら、入ろうとする側が待ちます。
  • 0.0秒に3人、0.3秒に3人。コンロが3つなので、6つの注文は2回に分かれて0.6秒かかります。誰が先にコンロを取るかは実行ごとに変わることがあります。
  • Semaphore(1) は Lock と同じく一度に一人です。ただし Lock は解放済みのロックをもう一度放すと RuntimeError になるのに対し、セマフォは release() を余分に呼んでも黙って番号札が増えます(下のBoundedSemaphore)。
  • 実務では「同時に接続するコネクションは5つまで」「同時にダウンロードするファイルは3つまで」のように、一度に押し寄せると困る資源の数を制限するときに使います。

下のタイムラインでコンロの数を変えてみてください。黄色のバーはコンロを待っている時間、オレンジのバーは実際に調理している時間です。

厨房タイムライン

注文

6

料理人

3

コンロ

2

調理

待ち

キャンセル

失敗

料理人6人、注文6つ、コンロ3つ。コンロを1つに減らすと黄色の待ちバーが階段のように長くなり、6つに増やすと待ちバーがほとんど消えます。
  • コンロ3つ:3列はすぐにオレンジのバー、残りの3列は0.3秒のあいだ黄色で待ってからオレンジに変わります。全体で約0.6秒。
  • コンロ1つ:料理人が6人いても一人ずつ順番に調理します。最後の料理人は1.5秒待ち、全体で約1.8秒。一人でやるのと同じです。
  • コンロ6つ:待ちはほとんどなく約0.3秒。コンロが料理人の数だけあれば、それ以上は速くなりません。
  • 料理人を増やしてもコンロがそのままなら、増えた人数のぶん黄色のバーが増えるだけです。ボトルネックは人数ではなく、コンロの数です。

BoundedSemaphore:番号札を余分に掛けてしまうミスをつかまえる

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 times

BoundedSemaphore は、最初に決めた数より多く返されると ValueError を出します。数を制限するためにセマフォを使うなら、ほとんどいつでも BoundedSemaphore のほうが向いています。ミスが黙って見過ごされず、その場で爆発してくれるからです。

シェフのひとこと

セマフォは数を数えてくれるだけで、黒板を守ってくれるわけじゃない。コンロ3つに3人立ってても、その3人が同じ黒板に数を書いたらまた競合状態。3人まで入れる部屋の中で共有の値に触るなら、その中でもう一度Lockを使って。

4-2. 「オープン!」:Event (9分)

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
# True

wait() に時間の上限を渡すこともできます。このとき返ってくる値が大事です。合図がついていれば 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

調理

待ち

キャンセル

失敗

料理人たちは始まるとすぐ Event.wait() で待ち、0.5秒後に合図がつくと全員そろって調理を始めます。料理人の数を変えても、スタートの時刻は同じです。
  • どの列も0秒から0.5秒まで黄色で待ち、0.5秒にいっせいにオレンジへ変わります。全体で約0.8秒。
  • 料理人が1人でも8人でも、スタートラインは同じ0.5秒です。set() 一回が、待っている人を全員起こすからです。

4-3. 「材料が届いたら起こして」:Condition (10分)

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 up
  • with 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. 団体席の4皿はいっせいに:Barrier (5分)

団体席の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')]
  • ミンホは0.1秒で仕上げましたが、4人ともいちばん遅いタクさんが着いた0.4秒に一緒に通過します。
  • wait() は0からn-1まで、人ごとに違う番号を返します。CPythonでは到着した順に0、1、2、3です。この番号で「一人だけがやる仕事」(例:if i == 0: でログを残す)を決められます。
  • action= に渡した関数は、最後の一人が到着したときに一度だけ実行されます。皿を送り出す合図です。
  • Barrierは全員が通過すると最初の状態に戻り、次のテーブルにもまた使えます。wait(timeout=) が時間切れになるとBarrierが壊れ、待っていた全員が BrokenBarrierError を受け取ります。

4-5. 今夜の道具のまとめ (4分)

道具厨房でいえばいつ使う
Lockチョークひとつ共有の値を一度に一人だけが書き直すとき(3限目)
Semaphore(n)、BoundedSemaphore(n)コンロ前の番号札n枚同時にn個までしか使わせないとき
Event壁の信号灯「スタート」「終わり」のような一度きりの合図を全員に
Conditionロック + 「起こして」名簿「この条件になったら起こして」。共有の状態が変わるのを待つとき
Barrier(n)団体席のパスn人がそろったら一緒に出発

5つとも threading にあり、どれも3限目のロックの上に建てた道具です。実際、Semaphore、Event、Barrier は内部で Condition を使っていて、Condition は中にロック(指定しなければ RLock)をひとつ持っています。

閉店後 (10分)

今夜の目標は、オープンの合図の前には誰も火をつけず、コンロは同時に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.90s
  • 0.3秒までは誰も調理しません。最初の皿は0.6秒、最後の皿は0.9秒に出て、目標の1秒以内に収まります。
  • peak は何度回しても3です。0.6秒に出る3つの注文がどの番号かは、実行ごとに変わることがあります。
  • on_fire と peak は6人の料理人が一緒に書き直す値なので、lock で包みました。セマフォの中であってもそうです。

閉店間際、窓際の席に一人で座った女性客が、麺を一杯きれいに平らげます。腕時計にちらりと目をやり、小さな手帳に何かを書きつけると、静かに会計を済ませて出ていきました。

シェフがコンロの元栓をひとつずつ閉めながら言います。「今日はみんな、ちゃんと並べたね。でもコンロじゃなくて、包丁とまな板もひとつずつしかなかったら? ミンホが包丁を持ってまな板を待ってて、スアがまな板を持って包丁を待ってたら、どうなると思う?」

営業日誌 — 四日目の夜。料理人6人、コンロ3つ。オープンの合図(Event)の前は待機、コンロは番号札3枚(Semaphore)で制限し、0.9秒で6件完了。窓際の席にお客さんが一人、麺を一杯。

練習問題

  1. コンロのタイムラインで料理人6人、注文6つにして、コンロを1から6まで変えながら全体の時間を表にまとめてみましょう。コンロ4つと5つで時間が同じなのはなぜでしょう?
  2. 4-2の「営業終了」の合図の形を自分で作ってみましょう。closing = threading.Event() を用意し、料理人スレッドが0.1秒ごとに1皿ずつ作りながら、closing.is_set() なら止まるようにします。メインスレッドは0.35秒後に closing.set() を呼んで join() します。何皿できたでしょう?
  3. 4-4のBarrierの例で 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。道具は重ねて使う。

← 前へ

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

次へ →

5限目 · 包丁とまな板:デッドロック

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

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

収集項目:メールアドレス・目的:この講座の新しい授業・公開のお知らせ・保有期間:解除のご依頼まで。他の目的には使わず、第三者にも提供しません。お知らせメールに返信いただければすぐに削除します。

質問と回答

この回について気になることを書き込んでください。すべての言語のページの質問がここに集まります。

リンクや書式なしで、そのままの文字で表示されます。

R

ReflexKR

触って学ぶPythonとReflex。すべての例はReflex 0.10.0で実際に動かして確認しています。

ReflexKRはバナー広告なし・無料で運営しています。役に立ったら、コーヒー一杯で応援してください。 ☕ コーヒーで応援する

アフィリエイトと支援について
한국어English日本語简体中文EspañolDeutsch

© 2026 ReflexKR · Since 2026 · CC BY-NC 4.0

Built with Reflex