共有データの競合と排他制御
二つの更新が失われる実行順を再現し、ロックと比較交換の保護範囲を確かめます。
説明
カウンタへ1を加える処理は、値を読み、足し算をし、書き戻すという複数の操作に分かれることがあります。二つのスレッドが同じ値を先に読むと、それぞれ更新しても最終値が1回分しか増えない場合があります。
まず二つのスレッドが両方読み終えてから書く順序を作り、更新喪失を再現します。次にロックで読み書き全体を保護します。最後に比較交換を小さな状態モデルで試し、読み取った値が古くなっていた場合の再試行を確認します。
1読取りと書戻しの間に入る別の更新
最初に一つの共有値を二つの更新者が読む順序を、変数aとbで表します。両方が0を読んでから、それぞれ1を加えた値を書き戻します。
import threading後続のスレッドAPI; value = 0; a = value; b = value両方が同じ値を読む; value = a+1; value = b+1古い値からそれぞれ書き戻す; print(value)残った値結果は1です。aとbがそれぞれ保持する古い値は0なので、後の書込みは先の1を同じ1で上書きします。次にスレッドでもこの順序を再現します。
2更新喪失を再現する処理
sharedに0を入れ、二つのスレッドで同じworkerを実行します。Barrier(2)は両方の読取りが済むまで書戻しを待たせる同期点で、joinは各スレッドの終了を待ちます。
def race():再現用の関数
shared = [0]; gate = threading.Barrier(2)共有値と同期点
def worker():各スレッドの処理
old = shared[0]; gate.wait(); shared[0] = old + 1両方が読んでから書く
threads = [threading.Thread(target=worker) for _ in range(2)]二つのスレッド
for t in threads: t.start()実行開始
for t in threads: t.join()終了を待つ
return shared[0]最終値両方がold=0を保持してから、1を書きます。二回の書込みが直列でも、計算に使った読取り値が同じなら結果は1です。スレッド終了を待つjoinだけでは、その内部の更新処理は排他されません。
3二回加算したはずの結果
再現用の関数を呼びます。スレッドの実行順をバリアで制約しているため、同じ更新喪失を確認できます。
print(race())二つの更新後の値結果は1です。二回の加算を反映した期待値2より小さくなっています。GILの有無や特定の式の実装を根拠に安全だと決めず、共有値の読取りから書戻しまでに割込みが入る可能性を検討します。
4読み書き全体をロックで保護
同じ加算をLockのwithブロック内へまとめます。片方が読んで書き終えるまで、もう片方はブロックへ入れません。
def locked():排他付きの加算
shared = [0]; lock = threading.Lock()共有値とロック
def worker():各スレッドの処理
with lock: shared[0] = shared[0] + 1読取り・加算・書込みをまとめて保護
threads = [threading.Thread(target=worker) for _ in range(2)]二つのスレッド
for t in threads: t.start()実行開始
for t in threads: t.join()終了を待つ
return shared[0]最終値保護する範囲には、古い値の読取りも含めます。書込みだけをロックしても、二つのスレッドが事前に同じ値を読んでいれば更新喪失を防げません。
5排他した結果と比較
保護の有無で結果を並べます。入力はどちらも0、更新回数は2です。
print(race(),locked())同じ条件で比較排他なしは1、ありは2です。ロックを持つ間だけがクリティカルセクションであり、同じ共有値に触る全参加者が同じロックの規則に従う必要があります。
6比較交換で古い値を検出
CASは現在値が期待値と一致するときだけ新しい値へ交換する原子的操作です。ここでは一操作として呼び出すモデルを定義します。Python関数自体をハードウェアatomicとして使うものではありません。
cell = [0]比較交換の対象
二つの更新者がともに0を読んだ状態から始めます。次の比較交換では、先に成功した書込みを後の更新者が上書きするかどうかが変わります。
7同じ期待値で二回交換する
一操作としてのCASをモデル化します。一致しなければ値を変えずFalseを返します。
def cas(expected,new):期待値と新しい値
if cell[0]!=expected: return False変更されていたら失敗
cell[0] = new; return True一致したら交換実際の原子的な比較交換は、比較と更新の間に他の更新が入らない保証を持ちます。関数の行数が短いことだけでその保証が得られるわけではありません。
8失敗した更新を読み直して再試行
一人目は0→1に成功し、二人目の0→1は失敗します。二人目が新しい値を読み直してから1→2を試します。
print(cas(0,1),cas(0,1),cell[0])同じ期待値の競合; current = cell[0]現在値を読み直す; print(cas(current,current+1),cell[0])新しい値から再試行最初はTrue・False・1、再試行後はTrue・2です。CASによる更新は競合時に再計算が必要です。また複数の変数をまとめた不変条件は、一つのカウンタをatomicにするだけでは保護できない場合があります。
まとめ
更新喪失は二つの更新者が同じ古い値から書き戻す順序で生じます。
ロックは読取りから書戻しまで、守りたい条件に必要な範囲をまとめて保護します。
CASは競合を失敗として検出でき、更新の再試行や保護する条件の検討が必要です。
操作履歴と解説を振り返れます。
Python 3
import threading; value = 0; a = value; b = value; value = a+1; value = b+1; print(value)1
def race():
shared = [0]; gate = threading.Barrier(2)
def worker():
old = shared[0]; gate.wait(); shared[0] = old + 1
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads: t.start()
for t in threads: t.join()
return shared[0]print(race())1
def locked():
shared = [0]; lock = threading.Lock()
def worker():
with lock: shared[0] = shared[0] + 1
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads: t.start()
for t in threads: t.join()
return shared[0]print(race(),locked())1 2
cell = [0]
def cas(expected,new):
if cell[0]!=expected: return False
cell[0] = new; return Trueprint(cas(0,1),cas(0,1),cell[0]); current = cell[0]; print(cas(current,current+1),cell[0])True False 1 True 2
演習完了