Gossip プロトコル

Gossip は、ノードが周期的に少数のピアを選択して状態を交換し、更新が複数のローカルな相互作用を経て拡散する、対等な情報伝播手法の一種です。これにより、固定されたブロードキャストセンターへの依存が軽減されますが、ノードは同時に完全に同一の情報を参照するわけではありません。

したがって、Gossip はメンバーシップ、状態の要約、または収束可能な更新の伝播に適しています。しかし、それだけでは唯一のリーダーを承認したり、すべてのノードが同じ順序でコマンドを処理することを保証したりすることはできません。伝播とコンセンサスは異なる問題を解決します。

2つの分類を混同しない

Push、Pull、Push-Pull は情報の流れを記述するものであり、Rumor Mongering（噂流布）と Anti-Entropy（非熵化/整合性維持）は伝播と追従補完の戦略を記述するものです。これらは独立して組み合わせ可能な 2 つの次元です。

Rumor は通常、コストを抑えるために新しい更新を有限の回数だけ伝播させた後に停止します。Anti-Entropy は定期的に状態を比較し、偶然に漏れた部分を補完します。Pull によって 1 回の通信で追いつくのは、ピアが必要なデータを持っておらず、通信が成功し、容量が十分である場合に限られます。

なぜ初期の伝播が速いのか

理想的なモデルでは、更新を知っているノードが未知のノードと接触し続け、初期段階ではカバーされるノード数が急速に増加します。しかし後期になると、すでに更新を知っているノードをより多く選択するようになり、残りの少数のノードがカバーの末尾（テール）となります。

ランダムな混合が良好で、メッセージに十分な容量があるなどの仮定の下では、多くのエピデミック（伝染）プロトコルにおいて、伝播時間はノード数に対して対数的に増加します。これは確率とモデル条件に基づく結論であり、任意のネットワークが正確に ${\log }_{2}N$ ラウンドですべてのノードに到達できるという意味ではありません。

2つのデータセンター間で接続が切断されると、ローカルな拡散がどれだけ速くても、パーティションを跨ぐことはできません。また、更新の発生速度が利用可能な伝播帯域幅を超えると、バックログ（溜まり）が増加し続ける可能性があります。Efficient Reconciliation and Flow Control では、有限な容量下でのアンチエントロピーの制約について直接的に議論しています。

4ノード、2ノードが知情：次のラウンドですべてがカバーされるとは限らない

同期ラウンドの開始時に、ノード 0 と 1 のみが更新を知っていると仮定します。各知情ノードは、他の 3 ノードから等確率で 1 つのターゲットを選択して Push し、メッセージはすべて到達するとします。新しく知情となったノードは、次のラウンドになってから転送を行います。2つの送信者が持つ 3×3=9 通りの等確率の選択を列挙できます。

from itertools import product
from fractions import Fraction

known = {0, 1}
outcomes = [known | {target0, target1}
            for target0, target1 in product((1, 2, 3), (0, 2, 3))]
assert len(outcomes) == 9
assert sum(len(result) == 4 for result in outcomes) == 2
assert sum(len(result) == 2 for result in outcomes) == 1
expected_new = Fraction(sum(len(result) - 2 for result in outcomes), 9)
assert expected_new == Fraction(10, 9)

# 未知情のノードについて、各送信者がそれを選ばない確率は 2/3 です。
assert expected_new == 2 * (1 - Fraction(2, 3)**2)

このうち、0→2,1→3 と 0→3,1→2 のみで 4 ノードすべてをカバーでき、その確率は 2/9 です。0→1,1→0 の場合は新規カバーが全く増えず、確率は 1/9 です。その他の選択では、知情ノードが 1 つだけ増加します。平均して 10/9 ノードが増加し、機械的な倍増（2つ増加）が要求されるものではありません。

N ノード、k 個の知情ノードに一般化すると、上記の独立した一様選択かつパケット損失なしの条件下では、次のラウンドで新規に知情となるノード数の期待値は (N−k)×[1−(1−1/(N−1))^k] となります。これは 1 ラウンドあたりの期待値であり、実際の各伝播軌跡ではありません。ネットワークパーティション、ローカル志向の選択、またはメッセージ容量の不足は、このモデルを変更します。

有限回の伝播では、確率的な漏れが残る可能性があります。例えば、A が更新を B に伝えた後に停止し、B が唯一の接触で A を選んだ場合、C や D はその Rumor を二度と受信できません。周期的なアンチエントロピーの意義は、後続の追従機会を提供することにあります。もし更新自体が完全に破棄されていれば、後続の接触がどれだけ頻繁でも、内容は回復できません。

2ノードが知情でも、次のラウンドで 4 になるとは限らない

期待値の式は、未知のノード 1 つごとに分解して理解できます。1 つの送信者が特定の未知ノードを選ばない確率は $1-1/(N-1)$ です。k 個の独立した送信者がすべてそれを選ばない確率は、この数の k 乗です。補集合を取ることで、そのノードがこのラウンドで知情となる確率が得られ、これを N-k 個の未知ノードについて合計します。異なる未知ノードが知情となるかどうかは相互に独立である必要はなく、期待値の線形性（可加性）だけで十分です。

$$E[\Delta ]=(N-k)\left[1-{\left(1-\frac{1}{N-1}\right)}^k\right]$$

重複排除、バージョン、マージが最終状態を決定する

同じ更新が異なる経路で複数回到達する可能性があります。受信側は、副作用の重複実行を避けるために、安定した識別子またはバージョンルールが必要です。もし完全な状態が伝播される場合は、2つの状態をどのようにマージするかを定義する必要があります。もし操作が伝播される場合は、消失、重複、順序を考慮する必要があります。

例えば、各ノードが独自の単調増加カウンタコンポーネントを維持し、別の副本の状態を受信した後に、各コンポーネントについて最大値を取り、合計することで、重複メッセージによるカウンタの重複累算を防ぐことができます。「合計を 1 増やす」という指示を直接ブロードキャストし、受信するたびに実行すると、重複に対して耐性を持ちません。

新旧の判断は、ローカルでの受信時刻だけでは行えません。遅延メッセージは新しいメッセージよりも遅く到着する可能性があります。メンバーが再起動した際にカウンタが 0 にリセットされる場合、incarnation（生成）などの世代情報を必要とします。より一般的なマージ条件については、衝突解決 を参照してください。

G-Counter：各ライターのカプセル状態を交換し、スナップショットを再度追加しない

3人のライターがそれぞれ自分のコンポーネントのみを増加させます。A は 2 回増加し、B は 3 回、C は 1 回増加しました。副本間で状態を交換する際、同じライターについては大きい方のコンポーネントを取り、合計します。重複したスナップショットは新しい操作にはなりません。

from itertools import permutations

def join(a, b):
    return {writer: max(a.get(writer, 0), b.get(writer, 0))
            for writer in a.keys() | b.keys()}

updates = [{"A": 2}, {"B": 3}, {"C": 1}]
for order in permutations(updates):
    state = {}
    for update in (*order, order[0], order[1]):  # 意図的に重複させ、順序を変更
        state = join(state, update)
    assert state == {"A": 2, "B": 3, "C": 1}
    assert sum(state.values()) == 6
assert join({"A": 2}, {"A": 1}) == {"A": 2}  # 遅れた古い状態は後退しない

# 再起動後に身份を再利用してローカルの累積値を 0 にリセットすると、新しい増加が古いコンポーネントに飲み込まれてしまう。
assert join({"A": 2}, {"A": 1}) != {"A": 3}
# 新しい incarnation では新しい身份を使用できる；古いコンポーネントは保持される。
assert sum(join({"A#1": 2}, {"A#2": 1}).values()) == 3

コンポーネントごとの max 演算は、交換法則、結合法則、冪等性を満たすため、完全な状態が重複・乱序で到達しても収束します。前提条件は、すべての増加が最終的に伝播され、各ライターが自分の単調なコンポーネントを独占していることです。2つのプロセスが同じ身份を誤用してそれぞれ 0 から 1 まで増加した場合、マージすると 1 しか得られず、2回の独立した増加を表現できません。

再起動時に永続的なコンポーネントを復元するか、新しい incarnation を使用することができますが、新しい身份を使用することは無コストな修復ではありません。古いコンポーネントは随意に削除できず、そうすると合計値が減少するため、身份の回収や圧縮プロトコルを別途設計する必要があります。ここで示す G-Counter は、増加のみをサポートし、任意の減算はサポートしません。状態型 CRDT のマージ条件および GCounter の実装例については、Akka Distributed Data を参照してください。

メンバーシップ伝播とデータアンチエントロピー

SWIM のプローブ層は存続証拠を取得し、伝播層は参加、離脱、疑念、存続の更新を拡散します。dead イベントのみを伝播として記述するのは不十分です。疑念の反論や新規メンバーの参加も、同様に拡散する必要があります。プロトコルの詳細については SWIM を参照してください。

大規模な業務データの場合、完全な状態を逐条伝播するのは高コストになり得るため、まず要約を交換し、同期が必要な範囲を特定します。Merkle tree はそのためのツールの一つであり、伝送を削減するのに役立ちますが、バージョンの衝突を自動的に解決するわけではありません。詳細は アンチエントロピーとデータ修復 を参照してください。

Fanout、周期、ペイロード予算

ノードあたり 1 秒間に 3 人のピアと連絡し、1 回の送信で約 2 KB を送信する場合、1 ノードあたりの単方向送信量は約 6 KB/s です。1000 ノードでは約 6 MB/s となり、返信やプロトコルオーバーヘッドは含まれていません。ここで示しているのは推計手法であり、任意の実装のデフォルト設定ではありません。

Fanout を高めたり周期を短くしたりすると、伝播が速くなる一方で、輻輳や処理待ちキューが発生し、実際の収束が逆に遅くなる可能性があります。単一の更新の再送信回数、メッセージサイズ、キューのバックログを制限し、アンチエントロピーのための追従機会を確保する必要があります。

障害関連の更新は、優先度の低い統計情報よりも緊急性が高いですが、優先度付けによって特定のカテゴリが永久に送信されなくなることはあってはなりません。周期的な要約とイベントの高速伝播を組み合わせることで、即時性と最終的なカバレッジの両立を図ります。

Gossip で代替できないもの

2つのノードが一時的に異なるメンバーシップリストを保持していることは、Gossip が許容する状態です。もしそれぞれがローカルのリストに基づいて排他的な書き込み権限を付与すると、同時に自分自身が唯一の所有者であるとみなす可能性があります。排他的な意思決定が必要な場合は、弱い一貫性を持つビューを、明確なセキュリティルールを持つ調整プロトコルに接続する必要があります。

同様に、Raft の AppendEntries は状態マシンレプリケーションプロトコルの一部であり、それ自体がすべての信頼できるブロードキャストの定義として使用できるわけではありません。システムを判断する際は、伝播、順序付け、永続化、配信保証をそれぞれ明確に記述し、「リーダーがいるかどうか」だけで正しさを比較しないようにしてください。

参考文献

• Efficient Reconciliation and Flow Control for Anti-Entropy Protocols
• SWIM 論文
• 障害検出

キーワード：gossip、push、pull、push-pull、rumor mongering、anti-entropy、fanout、確率的カバレッジ、状態マージ