最終課題：ロックフリーな市場データのファンアウトと不変条件の証明

読了 5分コード例 3件

最終プロジェクトはファンアウトです。1つのプロデューサがSPYのティックを公開し、複数のコンシューマがそのすべてを、ロックなしで読みます。これまで作ったものがここで集まります。プロデューサだけが動かすテール、誰も競合しないようにコンシューマごとに持つ私的なカーソル、リーダーに何が見えるかを決める書き込み順序、そして「動いているようだ」をすべてのスケジュールについての主張に変える探索器です。

設計：ログと、それぞれのカーソル

フィードは増える一方のログです。位置 p は tick<p> を持ち、tail は公開済みの位置の数を表します。ティックとテールを書くのはプロデューサだけなので、プロデューサ側に競合はありません。各コンシューマは自分のカーソル（次に読む位置）を持ち、他の誰もそれを書きません。これがファンアウトの要点です。コンシューマは他のコンシューマが読むものを一切書かないので、コンシューマを増やしても他に負担がかかりません。共有される事実はテールだけで、問いはプロデューサがティックを書くのとテールを動かすのとどちらが先かだけです。データのあとに動かせば、新しいテールを見たリーダーはデータを見つけます。前に動かせば、リーダーはまだ書かれていない位置を見ることがあります。

main.pyPython

def publish(shared, tick, safe=True):    if safe:        yield        pos = shared["tail"]        shared["tick%d" % pos] = tick        shared["put%d" % pos] = tick        yield        shared["tail"] = pos + 1    else:        yield        pos = shared["tail"]        shared["tail"] = pos + 1        yield        shared["tick%d" % pos] = tick        shared["put%d" % pos] = tickfor safe in (True, False):    shared, _ = run_prefix(lambda sh: [publish(sh, 9369, safe)], [0], make_feed(1))    print("safe  " if safe else "unsafe", "after one turn:", {k: v for k, v in shared.items() if k in ("tail", "tick0")})

safe   after one turn: {'tail': 0, 'tick0': 9369}
unsafe after one turn: {'tail': 1}

コンシューマ：テールを読み、次に自分のカーソル

ポーリングは2ターンです。最初のターンでテールを読みます。2番目のターンで自分のカーソルを読み、カーソルがテールより後ろならティックを読み、カーソルを進め、得たものを記録します。このコンシューマだけが自分のカーソルを書くので、位置が足元で変わることはありません。問題が起きうるのはティックの中身です。テールが動いた後、データが届く前に入ったリーダーは何も読めません。ポーリングはそれを None として記録するので、検査器が見つけられます。

main.pyPython

def poll(shared, c):    yield    tail = shared["tail"]    yield    pos = shared["cur%d" % c]    if pos >= tail:        return None    value = shared.get("tick%d" % pos)    shared["cur%d" % c] = pos + 1    shared["got%d_%d" % (c, pos)] = value    return valueready = dict(make_feed(1), tail=1)shared, alive = run_prefix(lambda sh: [poll(sh, 0)], [0, 0], ready)print("tail is 1 but tick0 was never written:", {k: v for k, v in shared.items() if k.startswith(("cur", "got"))})

tail is 1 but tick0 was never written: {'cur0': 1, 'got0_0': None}

コンシューマが単に待てない理由

何も見つからないポーリングは、ティックが現れるまで回り続けず、すぐに None を返します。これで各スレッドのプログラムが有限になり、探索器が終了できます。同時にこれはロックフリーの契約でもあります。どのスレッドも、自分の呼び出しを終えるために他のスレッドの進行に依存しません。

3つの不変条件、そしてすべてのスケジュール

フィードにとって「正しい」とは何でしょうか。最終状態についての3つの主張があります。Torn：コンシューマが受け取ったティックはすべて、その位置に置かれたティックと等しい（欠けたティックが None として受信されたものはtorn）。Gap：各コンシューマは位置0, 1, 2, ... を穴なく受け取る。Overrun：テールを超えるカーソルがない。verify はプロデューサとコンシューマのすべてのスケジュールを探索し、それぞれを壊すスケジュールを数えます。探索器が証明です。運の良い1回の実行をサンプリングするのではなく、すべてを訪れます。

main.pyPython

def violations(final):    puts = {int(key[3:]): value for key, value in final.items() if key.startswith("put")}    gots = {}    for key, value in final.items():        if key.startswith("got"):            c, p = key[3:].split("_")            gots[(int(c), int(p))] = value    broken = []    if any(value != puts.get(p) for (c, p), value in gots.items()):        broken.append("torn")    for c in {c for c, _ in gots}:        if sorted(p for cc, p in gots if cc == c) != list(range(sum(1 for cc, _ in gots if cc == c))):            broken.append("gap")            break    if any(key.startswith("cur") and value > final["tail"] for key, value in final.items()):        broken.append("overrun")    return sorted(broken)def verify(ticks, polls, safe=True):    outcomes = explore_all(feed_threads(ticks, polls, safe), make_feed(len(polls)))    broken = {}    delivered = 0    for key, count in outcomes.items():        final = dict(key)        delivered = max(delivered, sum(1 for k in final if k.startswith("got")))        for name in violations(final):            broken[name] = broken.get(name, 0) + count    return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "delivered": delivered}ticks = [9369, 9305]print("safe  ", verify(ticks, (2, 1), True))print("unsafe", verify(ticks, (2, 1), False))

safe   {'schedules': 3150, 'broken': {}, 'delivered': 3}
unsafe {'schedules': 3150, 'broken': {'torn': 869}, 'delivered': 3}

実データが示すものSPYの最初の2つの終値（9,369と9,305セント）を持つプロデューサと、2回ポーリングするコンシューマ、1回ポーリングするコンシューマ

スケジュール3,150各順序で探索

データが先0スケジュールが不変条件を壊す

テールが先869スケジュールが存在しないティックを渡す

穏やかな実行2各コンシューマが順に受け取るティック

同じプログラム、3,150通りのスケジュールで、2行の順序を入れ替えただけで「壊れたものなし」が869通りの破損に変わります。その多くは、数千回のテスト実行ではほとんど当たらないスケジュールです。これがテストの代わりに探索する理由であり、スキル全体が1つの表に収まっています。列挙できるほど小さなモデル、不変条件の検査器、そしてすべてのスケジュールについての判定です。

これから作るもの

ラボは講義の順に進みます。空のフィード、2つの順序のプロデューサ、コンシューマのポーリング、3つの不変条件の検査器、全スケジュールを探索する検証器、各コンシューマがティックを順に受け取る穏やかな実行の再生、そしてSPYの終値での研究です。

関数 · 問い前提
make_feed(consumers)何が共有されるか？単純な算術
publish(shared, tick, safe)データ先かテール先か単純な算術
poll(shared, c)コンシューマcの次のティック単純な算術
violations(final)どの不変条件が壊れたか？単純な算術
verify(ticks, polls, safe)フィードは全スケジュールで正しいか？explore_allviolations
drain(ticks, polls)各コンシューマは何を受け取るか？run_prefix
study(bars)SPYでの安全、危険、穏やかverifydrain

クイックチェック

危険な順序ではプロデューサが先にテールを動かします。どのリーダーがtornなティックを見ますか。

答えを選んでください。

これでロックフリーのスキルは完了です。部品（アトミック、compare-and-swap、スタックとキュー）を作り、危険（競合、ABA、回収）を名指しし、何よりたまたま通った1回の実行を信じる代わりに、設計をすべてのスケジュールに照らして確かめられるようになりました。次の段階は、同じ規律を実機で行うC++のアトミックのトピックです。

ラボ：採点される内容

7タスク · 100点

空のフィード5点
プロデューサ15点
コンシューマ20点
不変条件15点
検証器15点
穏やかな実行10点
SPYの表での研究20点

def publish(shared, tick, safe=True): if safe: yield pos = shared["tail"] shared["tick%d" % pos] = tick shared["put%d" % pos] = tick yield shared["tail"] = pos + 1 else: yield pos = shared["tail"] shared["tail"] = pos + 1 yield shared["tick%d" % pos] = tick shared["put%d" % pos] = tickfor safe in (True, False): shared, _ = run_prefix(lambda sh: [publish(sh, 9369, safe)], [0], make_feed(1)) print("safe " if safe else "unsafe", "after one turn:", {k: v for k, v in shared.items() if k in ("tail", "tick0")})

def poll(shared, c): yield tail = shared["tail"] yield pos = shared["cur%d" % c] if pos >= tail: return None value = shared.get("tick%d" % pos) shared["cur%d" % c] = pos + 1 shared["got%d_%d" % (c, pos)] = value return valueready = dict(make_feed(1), tail=1)shared, alive = run_prefix(lambda sh: [poll(sh, 0)], [0, 0], ready)print("tail is 1 but tick0 was never written:", {k: v for k, v in shared.items() if k.startswith(("cur", "got"))})

def violations(final): puts = {int(key[3:]): value for key, value in final.items() if key.startswith("put")} gots = {} for key, value in final.items(): if key.startswith("got"): c, p = key[3:].split("_") gots[(int(c), int(p))] = value broken = [] if any(value != puts.get(p) for (c, p), value in gots.items()): broken.append("torn") for c in {c for c, _ in gots}: if sorted(p for cc, p in gots if cc == c) != list(range(sum(1 for cc, _ in gots if cc == c))): broken.append("gap") break if any(key.startswith("cur") and value > final["tail"] for key, value in final.items()): broken.append("overrun") return sorted(broken)def verify(ticks, polls, safe=True): outcomes = explore_all(feed_threads(ticks, polls, safe), make_feed(len(polls))) broken = {} delivered = 0 for key, count in outcomes.items(): final = dict(key) delivered = max(delivered, sum(1 for k in final if k.startswith("got"))) for name in violations(final): broken[name] = broken.get(name, 0) + count return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "delivered": delivered}ticks = [9369, 9305]print("safe ", verify(ticks, (2, 1), True))print("unsafe", verify(ticks, (2, 1), False))