一覧OS-08-2 · 同期と資源競合

セマフォと条件変数

同時利用数を制限するセマフォと、共有状態の成立を待つ条件変数を操作します。

説明

ロックは一度に一つの処理だけを通すために使います。一方、二つまで使える接続枠のように残数を管理したい場合にはセマフォ、キューにデータが入るなどの状態変化を待ちたい場合には条件変数を使えます。

二枠のセマフォを取得・返却し、空きがない場合を確かめます。その後、条件変数で空のキューとデータの入ったキューを扱い、通知そのものと「取り出せるデータがある」という条件が別であることを確認します。

1二つの利用枠を取得

BoundedSemaphore(2)は初期値2のセマフォです。blocking=Falseで、空きがなければ待たずにFalseを返す取得を試します。

import threading同期API; slots = threading.BoundedSemaphore(2)二枠を用意; print(slots.acquire(blocking=False),slots.acquire(blocking=False),slots.acquire(blocking=False))三回の取得結果

True・True・Falseです。最初の二回で残数が0になり、三回目は枠を得られません。Falseだった処理が、後で枠を返すreleaseを呼んではいけません。

2返却すると次の利用者が入れる

一枠を返し、再度一枠を取得します。

slots.release()一枠を返却; print(slots.acquire(blocking=False))再取得

Trueになり、返した一枠を取得できました。ここでの残数は再び0です。セマフォは利用数を管理しますが、枠で守る実際の接続やバッファの内容を自動的に整合させるものではありません。

3過剰な返却を検出

残る二枠を返して初期状態へ戻します。BoundedSemaphoreは初期値を超えるreleaseをエラーにするので、次に過剰返却を試します。

slots.release(); slots.release()取得した二枠を返す

利用中の枠を両方返しました。ここからさらに返却すると、取得していない枠まで増やしたことになります。通常のSemaphoreは初期値を上限として強制しないため、資源プールではこのような検査が役立ちます。

4取得と返却の不一致

もう一回releaseし、過剰な返却を検出します。

try:余分な返却を試す slots.release()上限を超える返却 except ValueError:上限検査で失敗 print('release exceeds initial capacity')不一致を表示

上限超過としてValueErrorになりました。成功・失敗の分岐や例外処理があるプログラムでは、取得に成功した場合だけ、使い終わった後に必ず一度返す構造にします。

5キューが空かをロック内で確認

次は条件変数です。dequeをキューとして使い、Conditionが持つロックでキューの変更と確認を保護します。

from collections import dequeキュー; queue = deque()空から開始; cv = threading.Condition()状態確認用のロックを持つ条件変数; print(bool(queue))データがあるか

空なのでFalseです。待ちたい条件は「通知を受けた」ではなく「queueが空でない」です。条件変数はこの共有状態の確認と待機への移行を同じロックの規則でつなぎます。

6通知だけでは条件は成立しない

データを入れずにnotifyを呼び、その後wait_forでキューの状態を確認します。timeout=0は待ち時間を置かず、現在の条件だけを検査する指定です。

with cv:同じロックを保持 cv.notify()状態を変えずに通知; print(cv.wait_for(lambda: bool(queue),timeout=0))共有状態を検査

Falseです。通知はキューへデータを追加せず、後から来る待ち手のために回数を貯めるものでもありません。実際に待って目覚めた場合も、他の消費者が先にデータを取る可能性などがあるため、条件をもう一度確認します。

7データを追加してから通知

ロック内でjob-Aを追加して通知します。同じロックの規則に従う消費者が、空でないことを確かめてから取り出します。

with cv:キュー操作を保護 queue.append('job-A')データを追加; cv.notify()変更を通知; print(cv.wait_for(lambda: bool(queue),timeout=0))条件の成立; print(queue.popleft())成立を確認してから取出し

条件はTrueで、job-Aを取り出せます。notifyはロックを解放しません。別の待ち手が処理を進めるには、通知した側がwithを抜けるなどしてロックを解放し、待ち手が再取得する必要があります。

8別の消費者も同じ条件を確認する

先ほどの項目は取り出し済みです。次の消費者に相当する条件確認を行います。

with cv:同じロックの下で確認 print(cv.wait_for(lambda: bool(queue),timeout=0))次の消費者の条件; print(list(queue))空であることを照合

Falseで空のリストです。以前の通知だけを理由にpopleftすると空キューを取り出すことになります。通常の待機処理ではwhileで条件を再確認するか、wait_forのような再確認を行うAPIを使います。

まとめ

セマフォは利用枠の残数を管理し、成功した取得と返却を対応させます。

条件変数の通知と、処理を続けられる共有状態の成立は別です。

待機から戻った後も同じロックの下で条件を再確認し、成立した場合だけデータを取り出します。

操作履歴と解説を振り返れます。

Python 3

1
import threading; slots = threading.BoundedSemaphore(2); print(slots.acquire(blocking=False),slots.acquire(blocking=False),slots.acquire(blocking=False))
True True False
2
slots.release(); print(slots.acquire(blocking=False))
True
3
slots.release(); slots.release()
4
try:
 slots.release()
except ValueError:
 print('release exceeds initial capacity')
release exceeds initial capacity
5
from collections import deque; queue = deque(); cv = threading.Condition(); print(bool(queue))
False
6
with cv:
 cv.notify(); print(cv.wait_for(lambda: bool(queue),timeout=0))
False
7
with cv:
 queue.append('job-A'); cv.notify(); print(cv.wait_for(lambda: bool(queue),timeout=0)); print(queue.popleft())
True
job-A
8
with cv:
 print(cv.wait_for(lambda: bool(queue),timeout=0)); print(list(queue))
False
[]
演習完了