競合状態とロックが必要な理由

読了 7分コード例 4件

2つのスレッドが同じカウンタに1を足すと、エラーもクラッシュもなくカウンタが間違った値になります。x += 1 は1ステップではなく3ステップで、2つのスレッドがそれらをどんな順にインターリーブするかが答えを決めます。SPYの上昇日1,085日で542回の更新が消え、ロックを使えばすべて戻ります。

1行、3ステップ

プロセッサはメモリ上にある値に直接足し算できません。値をレジスタにロードし、レジスタに足し、レジスタをメモリにストアします。つまり x += 1 は load x、add 1、store x です。スレッドが1つなら3つを続けて行うので何も問題は起きません。スレッドが2つあると、マシンは片方の1ステップ、次にもう片方の1ステップを、任意の順で実行できます。これを確かめるために小さなマシンを書きます。スレッドはステップのリストで、それぞれ自分のレジスタを持ち、スレッド同士は名前付きの値の辞書を共有します。関数 execute はスレッドの1ステップを実行します。次に、2つのスレッドの6ステップを、交互に1ステップずつ手で動かしてみます。

main.pyPython

def execute(op, shared, regs, locks, tid):    kind = op[0]    if kind == "load":        regs[tid] = shared.get(op[1], 0)    elif kind == "add":        regs[tid] += op[1]    elif kind == "store":        shared[op[1]] = regs[tid]    elif kind == "lock":        if op[1] in locks:            return False        locks[op[1]] = tid    elif kind == "unlock":        del locks[op[1]]    return Trueshared, regs, locks = {"x": 0}, {0: 0, 1: 0}, {}turns = [(0, 0), (1, 0), (0, 1), (1, 1), (0, 2), (1, 2)]1for tid, step in turns:    execute(INCREMENT[step], shared, regs, locks, tid)    print("thread", tid, INCREMENT[step], "->", shared, regs)

thread 0 ('load', 'x') -> {'x': 0} {0: 0, 1: 0}
thread 1 ('load', 'x') -> {'x': 0} {0: 0, 1: 0}
thread 0 ('add', 1) -> {'x': 0} {0: 1, 1: 0}
thread 1 ('add', 1) -> {'x': 0} {0: 1, 1: 1}
thread 0 ('store', 'x') -> {'x': 1} {0: 1, 1: 1}
thread 1 ('store', 'x') -> {'x': 1} {0: 1, 1: 1}

1どちらかがストアする前に両方がロードするので、どのレジスタも0です。

出力を1行ずつ読みましょう。スレッド0は0をロードし、スレッド1も0をロードします（スレッド0はまだストアしていません）。どちらも0のレジスタに1を足し、どちらも1をストアします。2つのスレッドがそれぞれ x += 1 をしたのに、x は2ではなく1です。更新が1つ失われました。どちらのスレッドのコードにも誤りはなく、バグはこのステップの順序にだけ存在します。

定義競合状態

スレッドがステップをインターリーブする順序によって結果が変わるバグ。同じプログラムでも、あるスケジュールでは正しく別のスケジュールでは誤るため、テストで見逃しやすい。

スケジュールはスレッド番号のリスト

現実の世界では、スケジュールはOSが決め、実行のたびに変わるので、競合は再現しにくいのです。私たちのマシンでは自分で選びます。スケジュールはスレッド番号のリストで、run は各エントリが指すスレッドに手番を1回ずつ与えます。大事なのは2つの単純なスケジュールです。直列スケジュールではスレッド0が全ステップを実行してからスレッド1が実行します。これが私たちの望む形です。ラウンドロビンスケジュールではスレッドが1ステップずつ交互に進みます。2つのスレッドのステップがあらゆる場所で重なるので、このカウンタにとって最悪のケースです。関数 lost_updates は、増分が足すべきだった量とカウンタの値を比べます。

main.pyPython

def run(programs, schedule, shared=None):    shared = dict(shared or {})    regs = {tid: 0 for tid in range(len(programs))}    pcs = [0] * len(programs)    locks = {}    for tid in schedule:        if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid):            pcs[tid] += 1    while any(pcs[t] < len(programs[t]) for t in range(len(programs))):        progress = False        for tid in range(len(programs)):            if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid):                pcs[tid] += 1                progress = True        if not progress:            raise RuntimeError("deadlock")    return shareddef serial_schedule(programs):    return [tid for tid, program in enumerate(programs) for _ in program]def round_robin_schedule(programs):    longest = max((len(program) for program in programs), default=0)    return [tid for step in range(longest) for tid, program in enumerate(programs) if step < len(program)]def lost_updates(programs, schedule, name="x"):    expected = sum(op[1] for program in programs for op in program if op[0] == "add")    return expected - run(programs, schedule).get(name, 0)two = [INCREMENT, INCREMENT]print(serial_schedule(two), run(two, serial_schedule(two)))print(round_robin_schedule(two), run(two, round_robin_schedule(two)))print("lost:", lost_updates(two, round_robin_schedule(two)))

[0, 0, 0, 1, 1, 1] {'x': 2}
[0, 1, 0, 1, 0, 1] {'x': 1}
lost: 1

直列スケジュール [0, 0, 0, 1, 1, 1] は2、ラウンドロビン [0, 1, 0, 1, 0, 1] は1になり、更新が1つ失われます。run はスケジュールが早く終わった場合、未完了のスレッドにスレッド順で1ステップずつ手番を与え続けるので、どのプログラムも必ず終わります。また、完了したスレッドは手番を無視するので、どんなスレッド番号のリストも有効なスケジュールです。

ロックはステップを1つのブロックにする

ロックは、重なりを禁止することで競合を直します。スレッドは load の前にロックを取り、store の後に解放します。その間にロックを求めた2つ目のスレッドは取得できず、その手番は何もせず、ロックが空くまで同じステップに留まります。こうして lock と unlock の間のステップは1つのブロックとして実行され、他のスレッドは割り込めません。関数 counter_program は、カウンタを増やすスレッドのプログラムを、ロックありまたはなしで作ります。ロックには新しい危険があり、コードの最後に示されています。2つのロックを逆の順に取る2つのスレッドは、それぞれ相手が必要とするロックを握り、どちらも動けなくなります。これがデッドロックで、1周しても進展がなければ run が検出します。

main.pyPython

def counter_program(increments, locked=False):    program = []    for _ in range(increments):        program += ([("lock", "m")] if locked else []) + INCREMENT + ([("unlock", "m")] if locked else [])    return programsafe = [counter_program(1, locked=True), counter_program(1, locked=True)]print(counter_program(1, locked=True))print(run(safe, round_robin_schedule(safe)))ab = [[("lock", "a"), ("lock", "b"), ("unlock", "b"), ("unlock", "a")],      [("lock", "b"), ("lock", "a"), ("unlock", "a"), ("unlock", "b")]]try:    run(ab, [0, 1])except RuntimeError as error:    print("error:", error)

[('lock', 'm'), ('load', 'x'), ('add', 1), ('store', 'x'), ('unlock', 'm')]
{'x': 2}
error: deadlock

ロックはタダではない

ロックを待つスレッドは何もせず、ロックを持つスレッドはどのステップでもOSに一時停止されうるので、トレーディングのホットパスでは上限の決められないジッタになります。このスキルの残りは、ロックをまったく必要としない構造を作ります。

main.pyPython

def study(bars):    returns = [float(r) for r in bars["close"].pct_change().dropna()][:2000]    ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)]    free = [counter_program(k) for k in ups]    locked = [counter_program(k, locked=True) for k in ups]    return {"up": sum(ups),            "serial": run(free, serial_schedule(free))["x"],            "racy": run(free, round_robin_schedule(free))["x"],            "locked": run(locked, round_robin_schedule(locked))["x"]}print(study(load_bars()))

{'up': 1085, 'serial': 1085, 'racy': 543, 'locked': 1085}

実データが示すものSPYの日次リターン2,000件、2スレッドで上昇日を数える、ラウンドロビン

数えるべき上昇日1,0852,000日中

直列スケジュール1,085最終的な x。すべて数えられた

ラウンドロビン・ロックなし543最終的な x。542回の更新が消えた

ラウンドロビン・ロックあり1,085最終的な x。何も失われない

スケジュール別のカウンタ最終値

直列1085

ラウンドロビン・ロックなし543

ラウンドロビン・ロックあり1085

ロックがないと、ラウンドロビンでは2つのスレッドが常に1ステップ違いで進むため、いつも同じ値をロードし、増分は2回に1回しか数えられません。1,085日の上昇日のおよそ半分が失われます。ロックを使えば正確な合計が戻りますが、代わりにどちらかのスレッドが待たされます。

これから作るもの

ラボは講義の順に進みます。スレッドの1ステップ、スケジュールの下でのマシン全体、2つの単純なスケジュール、失われた更新の数、ロックありなしのカウンタ用スレッドのプログラム、そして2つのスレッドでSPYの上昇日を数える研究です。

関数 · 問い前提
execute(op, shared, regs, locks, tid)1ステップは何をするか、スレッドは待つか？単純な算術
run(programs, schedule, shared)このスケジュールでスレッドは何を残すか？execute
serial_schedule(programs)1つずつ順番に？単純な算術
round_robin_schedule(programs)交互に1ステップずつ？単純な算術
lost_updates(programs, schedule)増分は何回消えるか？run
counter_program(increments, locked)カウンタ用スレッドは何をするか？単純な算術
study(bars)SPYで競合は何回の更新を失わせるか？counter_programrunlost_updates

クイックチェック

2つのスレッドがそれぞれ x = 0 に対して load x、add 1、store x を、スケジュール [0, 1, 1, 0, 0, 1] で実行します。最終的な x はいくつですか。

答えを選んでください。

次のトピック：手でスケジュールを選んで、失われた更新を見つけました。実際のプログラムにはスケジュールが多すぎて、手では試せません。インターリーブ探索器では、小さなプログラムの全スケジュールを列挙して、壊れるものを数え、その探索器をこのスキルの残りの道具として使います。

ラボ：採点される内容

7タスク · 100点

スレッドの1ステップ15点
スケジュールの下でのマシン25点
1つずつ順番に5点
交互に1ステップずつ10点
失われた更新10点
カウンタ用スレッド10点
SPYの表での研究25点

def execute(op, shared, regs, locks, tid): kind = op[0] if kind == "load": regs[tid] = shared.get(op[1], 0) elif kind == "add": regs[tid] += op[1] elif kind == "store": shared[op[1]] = regs[tid] elif kind == "lock": if op[1] in locks: return False locks[op[1]] = tid elif kind == "unlock": del locks[op[1]] return Trueshared, regs, locks = {"x": 0}, {0: 0, 1: 0}, {}turns = [(0, 0), (1, 0), (0, 1), (1, 1), (0, 2), (1, 2)]for tid, step in turns: execute(INCREMENT[step], shared, regs, locks, tid) print("thread", tid, INCREMENT[step], "->", shared, regs)

thread 0 ('load', 'x') -> {'x': 0} {0: 0, 1: 0} thread 1 ('load', 'x') -> {'x': 0} {0: 0, 1: 0} thread 0 ('add', 1) -> {'x': 0} {0: 1, 1: 0} thread 1 ('add', 1) -> {'x': 0} {0: 1, 1: 1} thread 0 ('store', 'x') -> {'x': 1} {0: 1, 1: 1} thread 1 ('store', 'x') -> {'x': 1} {0: 1, 1: 1}

def run(programs, schedule, shared=None): shared = dict(shared or {}) regs = {tid: 0 for tid in range(len(programs))} pcs = [0] * len(programs) locks = {} for tid in schedule: if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid): pcs[tid] += 1 while any(pcs[t] < len(programs[t]) for t in range(len(programs))): progress = False for tid in range(len(programs)): if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid): pcs[tid] += 1 progress = True if not progress: raise RuntimeError("deadlock") return shareddef serial_schedule(programs): return [tid for tid, program in enumerate(programs) for _ in program]def round_robin_schedule(programs): longest = max((len(program) for program in programs), default=0) return [tid for step in range(longest) for tid, program in enumerate(programs) if step < len(program)]def lost_updates(programs, schedule, name="x"): expected = sum(op[1] for program in programs for op in program if op[0] == "add") return expected - run(programs, schedule).get(name, 0)two = [INCREMENT, INCREMENT]print(serial_schedule(two), run(two, serial_schedule(two)))print(round_robin_schedule(two), run(two, round_robin_schedule(two)))print("lost:", lost_updates(two, round_robin_schedule(two)))

def counter_program(increments, locked=False): program = [] for _ in range(increments): program += ([("lock", "m")] if locked else []) + INCREMENT + ([("unlock", "m")] if locked else []) return programsafe = [counter_program(1, locked=True), counter_program(1, locked=True)]print(counter_program(1, locked=True))print(run(safe, round_robin_schedule(safe)))ab = [[("lock", "a"), ("lock", "b"), ("unlock", "b"), ("unlock", "a")], [("lock", "b"), ("lock", "a"), ("unlock", "a"), ("unlock", "b")]]try: run(ab, [0, 1])except RuntimeError as error: print("error:", error)

def study(bars): returns = [float(r) for r in bars["close"].pct_change().dropna()][:2000] ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)] free = [counter_program(k) for k in ups] locked = [counter_program(k, locked=True) for k in ups] return {"up": sum(ups), "serial": run(free, serial_schedule(free))["x"], "racy": run(free, round_robin_schedule(free))["x"], "locked": run(locked, round_robin_schedule(locked))["x"]}print(study(load_bars()))