デッドロックとロックの取得順序
資源の所有者と待ち先から循環を見つけ、共通の取得順序と失敗時の解放を試します。
説明
二つの処理がそれぞれ別のロックを持ち、相手の持つロックを待つと、どちらも先へ進めなくなる場合があります。このようなデッドロックは、単にCPUが遅い状態とは異なり、待ち先と所有者の関係に原因があります。
AとBという二つのロックを逆順で要求する状態を表にし、待ち関係をたどります。次にロックの取得順をA→Bへ統一したスレッドを実行します。最後に、追加のロックを取得できないときに保持中のロックを解放する処理を確かめます。
1各ロックの所有者
T1がAを、T2がBを持つ状態を作ります。次にT1はB、T2はAを要求する予定です。
owner = {'A':'T1','B':'T2'}資源の所有者; request = {'T1':'B','T2':'A'}次の取得要求; print(owner); print(request)二つの関係どちらの要求先も他方が保持しています。ロックを持ったまま相手の解放を待つと、各スレッドの次の処理が進めなくなります。
2待ち先をスレッドへ置き換える
要求しているロック名からその所有者を引き、待つ側→待たれる側のグラフを作ります。
waits = {thread:owner[resource] for thread,resource in request.items()}要求先を所有者へ変換; print(waits)待ちグラフT1→T2とT2→T1です。ロックごとに一つだけ保持者が存在し、待っている間に解放しない条件なので、この循環によって両方が停止します。複数個ある資源ではグラフの循環だけで同じ判断をできない場合があります。
3二つの待ち先をたどる
T1の待ち先、そのスレッドの待ち先を順に求めます。この二スレッドの表では、出発点へ戻るかを二段で確認できます。
start = 'T1'出発点; route = [start,waits[start],waits[waits[start]]]所有者を二段たどる; print(route)経路; print(route[0]==route[-1])出発点へ戻るかT1→T2→T1と戻り、循環が見つかります。ロックを持ったまま次を待つこと、途中で強制的に奪えないこと、排他資源であること、循環することが重なっています。
4取得順を全参加者でそろえる
各スレッドの取得するロックを同じA→Bの順へ並べます。
ordered = {t:sorted(['A','B']) for t in ('T1','T2')}共通順序; print(ordered)取得する並び両方がAを先に取得するため、Aを取れなかった側はBを保持していない状態で待ちます。全参加者が同じ総順序を守れば、この二資源の循環待ちを作らずに済みます。一部の処理だけ順序を直しても他の経路が逆順なら原因が残ります。
5順序統一後の処理を実行
二つのスレッドがA→Bの順にLockを取得し、処理した回数を増やします。withを抜けると逆順に解放します。どちらが先にAを得るかには依存せず、両方の終了を待つ関数です。
def run_ordered():順序を統一した実験
import threadingロックAPI
a = threading.Lock(); b = threading.Lock(); done = []資源と完了記録
def worker():各スレッド
with a:まずA
with b: done.append('done')次にBを得て完了記録
threads = [threading.Thread(target=worker) for _ in range(2)]二つの処理
for t in threads: t.start()開始
for t in threads: t.join()終了を待つ
return len(done)完了数同じロック順序を共有し、保護区間の処理を終えてから解放します。次に戻り値で、二つとも完了できたことを確かめます。
6完了数と失敗時の解放
二つの完了を確認します。続いてA・Bがそれぞれ保持された状況をLockで表し、追加取得に失敗したらAを返す処理を試します。
import threading取得失敗の確認用API; print(run_ordered())二つの完了; a = threading.Lock(); b = threading.Lock()二つの資源; a.acquire(); b.acquire()別の保持者がいる状態を表す; got_b = b.acquire(blocking=False); print(got_b)待たずに追加取得; a.release(); print(a.locked())保持していたAを返却; b.release()残るBも解放完了数は2です。続くTrue二つは初期の取得成功、Falseは追加取得の失敗、最後のFalseはAが解放された状態です。取得失敗時に保持資源を返す方針でも循環を解消できますが、全員が同じタイミングで再試行すると進めないライブロックなど、再試行方針にも注意が必要です。
まとめ
所有者と要求先を別々に調べると、誰が誰を待っているかを特定できます。
同じ総順序でロックを取得する規則は、参加する全処理に適用します。
タイムアウトや取得失敗だけでは保持資源は自動解放されないため、解放と再試行の方針も必要です。
操作履歴と解説を振り返れます。
Python 3
owner = {'A':'T1','B':'T2'}; request = {'T1':'B','T2':'A'}; print(owner); print(request){'A': 'T1', 'B': 'T2'}
{'T1': 'B', 'T2': 'A'}waits = {thread:owner[resource] for thread,resource in request.items()}; print(waits){'T1': 'T2', 'T2': 'T1'}start = 'T1'; route = [start,waits[start],waits[waits[start]]]; print(route); print(route[0]==route[-1])['T1', 'T2', 'T1'] True
ordered = {t:sorted(['A','B']) for t in ('T1','T2')}; print(ordered){'T1': ['A', 'B'], 'T2': ['A', 'B']}def run_ordered():
import threading
a = threading.Lock(); b = threading.Lock(); done = []
def worker():
with a:
with b: done.append('done')
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads: t.start()
for t in threads: t.join()
return len(done)import threading; print(run_ordered()); a = threading.Lock(); b = threading.Lock(); a.acquire(); b.acquire(); got_b = b.acquire(blocking=False); print(got_b); a.release(); print(a.locked()); b.release()2 True True False False
演習完了