更新が失われるのは、読み出しと書き込みが間に隙間のある2つの手番だからです。アトミック操作はその隙間を閉じ、読み出し・変更・書き込みを分割不能な1ステップで行います。これでカウンタはすべてのスケジュールで正しくなり、スケジュールは28通りから4通りに減り、間違いはゼロです。
1手番、隙間なし
このマシンでは手番は1つの yield から次の yield までなので、2つの yield の間に書かれたものはインターリーブされません。したがって yield を含まない関数はアトミック操作として振る舞います。最も一般的なのはfetch-and-addで、古い値を読み、old + n を書き、old を返すまでを1手番で行います。これを呼ぶカウンタ用スレッドは、増分全体で yield が2つではなく1つになるので、2つのスレッドが同じ古い値を読むことはもうありません。
def fetch_add(shared, name, n): old = shared.get(name, 0) shared[name] = old + n return olddef atomic_increments(k): def thread(shared): for _ in range(k): yield fetch_add(shared, "x", 1) return threadtwo = lambda shared: [atomic_increments(1)(shared), atomic_increments(1)(shared)]for prefix in ([0, 1], [1, 0]): print(prefix, run_prefix(two, prefix))
[0, 1] ({'x': 2}, [])
[1, 0] ({'x': 2}, [])どちらの順序も {'x': 2} で終わり、生きているスレッドは残りません。増分1回が1手番なので、プログラムのスケジュールは [0, 1] と [1, 0] の2通りだけ（研究では増分1回と3回で4通り）で、更新を失うものはありません。
比較交換とリトライループ
fetch-and-add は足し算しかできません。それ以外には汎用の道具、比較交換（CAS）が必要です。パターンはループです。古い値を読み、そこから新しい値を計算し、その間に誰も値を変えていなければCASで書き込みます。CASが失敗したら誰かが先に書いたということなので、読み直してやり直します。次は2つのCASカウンタです。スレッド0が書く前にスレッド1が読むと、スレッド1のCASは1回失敗してリトライを数え、2回目で成功します。古い値での書き込みは拒否されるので、更新は失われません。
def compare_and_swap(shared, name, expected, new): if shared.get(name, 0) != expected: return False shared[name] = new return Truedef cas_increments(k): def thread(shared): for _ in range(k): while True: yield old = shared.get("x", 0) yield if compare_and_swap(shared, "x", old, old + 1): break fetch_add(shared, "retries", 1) return threadcas = lambda shared: [cas_increments(1)(shared), cas_increments(1)(shared)]print(run_prefix(cas, [0, 1, 0, 1]))print(run_prefix(cas, [0, 1, 0, 1, 1, 1]))
({'x': 1, 'retries': 1}, [1])
({'x': 2, 'retries': 1}, [])[0, 1, 0, 1] のあと、共有辞書は {'x': 1, 'retries': 1} で、スレッド1はまだ生きています。0を読みましたが、スレッド0がすでに1をストアしていたため、比較交換は拒否されたのです。スレッド1に次の2手番（読み直しと再度の交換）を与えると x = 2 で終わります。代償に注目してください。負けた側は仕事を2回しました。fetch-and-add は決してリトライしませんが、CASはスレッドがぶつかった回数だけリトライします。
手番の数が変わるときの探索
前のトピックの探索器は、各スレッドの手番数が固定の lengths だと仮定していました。リトライループはそれを崩します。競争に負けたスレッドは余分な手番を取り、その数はスケジュール次第です。解決策は、スケジュールを木として探索することです。手番の接頭辞から run_prefix がプログラムを再生して、まだ生きているスレッドを教えてくれます。いなければスケジュールは完成なので最終状態を数え、いれば各スレッドで枝分かれします。この深さ優先探索は完成したスケジュールをちょうど1回ずつ訪れます。そのあと values_of が全結果から共有名の値を読み出します。
def explore_all(make_threads, shared=None): outcomes = {} def visit(prefix): final, alive = run_prefix(make_threads, prefix, shared) if not alive: key = tuple(sorted(final.items())) outcomes[key] = outcomes.get(key, 0) + 1 return for tid in alive: visit(prefix + [tid]) visit([]) return outcomesdef values_of(outcomes, name): return sorted({dict(key).get(name, 0) for key in outcomes})print(explore_all(cas))racy = lambda shared: [increments(1)(shared), increments(1)(shared)]print(values_of(explore_all(racy), "x"), values_of(explore_all(cas), "x"))
{(('x', 2),): 2, (('retries', 1), ('x', 2)): 4}
[1, 2] [2]このマシンでは、すべての操作がすべてのスレッドから同時に、順序どおりに見えます。実際のプロセッサは通常の読み書きにそれを保証せず、アトミック操作でさえコア同士がキャッシュラインを奪い合います。次のトピックで最初の仮定を外します。
returns = [float(r) for r in load_bars()["close"].pct_change().dropna()][:6]ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)]for name, factory in (("racy", increments), ("atomic", atomic_increments), ("cas", cas_increments)): outcomes = explore_all(lambda shared, f=factory: [f(k)(shared) for k in ups]) print(name, sum(outcomes.values()), "schedules, x in", values_of(outcomes, "x"))
racy 28 schedules, x in [1, 2, 3, 4] atomic 4 schedules, x in [4] cas 144 schedules, x in [4]
どちらの修正もすべてのスケジュールで正しく、違いはコストです。fetch-and-add は増分あたり1手番で済み、4通りです。CASカウンタは衝突で手番が増えるため144通りで、最悪の場合スレッドは3回リトライします。これがロックフリーコードのトレードオフです。誰もロックを待ちませんが、各競争の勝者が敗者に仕事のやり直しを強います。
共有値を読み、そこから新しい値を計算し、分割不能な1ステップで書き戻す操作。fetch-and-add と比較交換が最もよく使う2つで、すべてのロックフリー構造はこれらから作られます。
これから作るもの
ラボは講義の順に進みます。fetch-and-add、比較交換、それぞれを使ったカウンタ用スレッド、手番数が変わるスレッドを扱う探索器、全結果から名前の値を読む補助関数、そしてSPYの上昇日に対する3種類のカウンタの研究です。
- 関数 · 問い前提
- fetch_add(shared, name, n)n を足す。前の値は？単純な算術
- compare_and_swap(shared, name, expected, new)値はまだ読んだ値のままか？単純な算術
- atomic_increments(k)隙間のないカウンタは？fetch_add
- cas_increments(k)失う代わりにリトライするカウンタは？compare_and_swapfetch_add
- explore_all(make_threads, shared)手番数によらず全結果は？run_prefix
- values_of(outcomes, name)この名前はどんな値を取るか？単純な算術
- study(bars)SPYで3つのカウンタはどう違うか？explore_allvalues_of
次のトピック：ここまでは、どのスレッドもすべての書き込みを即座に順序どおり見ると仮定してきました。実際のプロセッサはそれを保証しません。メモリモデル：acquireとreleaseでは、マシンにストアバッファを加え、保護しているデータより先にフラグが見えてしまう読み手を示し、最もコストの小さい順序付けで直します。
ラボ：採点される内容
7タスク · 100点- fetch-and-add5点
- 比較交換10点
- fetch-and-add のカウンタ10点
- リトライループのカウンタ20点
- 全スケジュールの探索20点
- 名前の値5点
- SPYの表での研究30点