コード層からイーサリアム P2P ネットワーク設計を深く理解する
序文
序文
この記事では、P2P ネットワークの開発の概要を説明するとともに、イーサリアムにおけるノード検出メカニズム、分散ハッシュ テーブル、ノード ルックアップ、ノード追加、ノード削除についてソース コードの観点から簡単に紹介します。 P2P ネットワークのセキュリティ設計の簡単な分析。
基本知識
基本知識
P2Pネットワーク
P2P ネットワークは従来の CS 構造とは異なり、P2P ネットワークでは各ノードがクライアントまたはサーバーのいずれかになることができ、ノード間の通信プロトコルは通常 Socket を通じて直接実装されます。
P2P テクノロジーの開発は、次の 4 つの開発段階を経て行われます。
1. 集中化:
これは、P2P ネットワーク モードで最も単純なルーティング方法です。つまり、他のすべてのノードのインデックス情報を保存するセントラル ノードがあり、インデックス情報には通常、ノード IP、ポート、ノード リソースなどが含まれます。ルーティングは構造がシンプルで実装しやすいのが特徴 簡単だがデメリットも明らか セントラルノードは全ノードの経路情報を保持する必要があるため、ノード規模が拡大し続けると性能ボトルネックが発生しやすい、単一ノード障害の問題もあります
2. 配布:
これは、中央ノードが削除され、P2P ノード間にランダムなネットワークが確立され、新しく追加されたノードと P2P ネットワーク内のノードとの間に接続チャネルがランダムに確立され、ランダムなトポロジ構造が形成されることを意味します。既存のノードをランダムに選択し、隣接関係を確立します。ノードが隣接ノードとの接続を確立した後、ネットワーク全体にノードの存在を知らせるためにネットワーク全体にブロードキャストする必要があります。"具体的なブロードキャスト手順は次のとおりです。ノードは最初に隣接ノードにブロードキャストし、隣接ノードがブロードキャスト メッセージを受信した後、自身の隣接ノードにブロードキャストを継続するなど、このブロードキャスト方法はとも呼ばれます。"フラッディングメカニズム
フラッディング メカニズムの問題は、制御性が低いことです。これには、主に 2 つの大きな問題があります。1 つは、フラッディング サイクルが形成されやすいことです。たとえば、ノード A がノード B にメッセージをノード C に送信し、ノード C がブロードキャストします。ノード C A に送信されると、これがサイクルを形成します。もう 1 つの問題は、メッセージ応答の下請けです。たとえば、ノード A が要求したいリソースが多くのノードによって所有されている場合、短期間に多数のノードが同時に応答メッセージをノード A に送信します。これにより、ノード A が即座にクラッシュする可能性があります。
フラッディング ループを排除する方法は、IP ネットワークのルーティング プロトコルにおけるフラッディング ブロードキャストの制御に学ぶことができます。1 つの方法は、クエリ メッセージごとに TTL 値を設定することです。フラッディング メッセージが転送されるたびに、TTL 値は 1 ずつ減らされます。ノードが受け入れるとき TTL が 0 の場合、メッセージは転送されません。これにより、ネットワーク内のクエリ メッセージの無限ループを回避でき、フラッド メッセージに固有のフラグを設定することもでき、受信した重複メッセージは転送されません。ストームに応答する方法では、データ リンク層でネットワーク セグメンテーションを実行して、セグメント間でのメッセージのブロードキャストを削減できます。
3. ハイブリッド:
ハイブリッド型とは、実際には集中型と分散型が混在した構造であり、ネットワーク内にはスーパーノードで構成される分散型ネットワークが多数存在し、各スーパーノードは複数の通常ノードを持ち、ローカルな集中型ネットワークを形成します。最初に新しい通常ノードを追加することができます。通信するスーパー ノードを選択すると、スーパー ノードは他のスーパー ノード リストを新しい参加ノードにプッシュします。参加ノードは、リスト内のスーパー ノードのステータスに応じて、特定のスーパー ノードを親ノードとして選択することを決定します。この構造のフラッド ブロードキャストはスーパー ノード間で発生するだけであり、大規模なフラッディングを回避できます。実際のアプリケーションでは、ハイブリッド構造は比較的柔軟で効果的なネットワーク アーキテクチャであり、実装の難易度は比較的低いです。そのため、現在多くのシステムが導入されています。ハイブリッド構造に基づいた開発と実装
4. 構造化:
構造化 P2P ネットワークは、分散型ネットワーク構造です。上記の分散型構造とは異なり、分散型ネットワークはランダムなネットワークですが、構造型ネットワークは、リング ネットワークやリング ネットワークを形成するなど、特定の構造に従ってすべてのノードを規則的に組織します。ツリー ネットワーク、構造化ネットワークは一般に、特定の実装では分散ハッシュ テーブル (分散ハッシュ テーブル、DHI) アルゴリズムに基づいており、特定の実装スキームには Chord、Pasty、CAN、Kademlia およびその他のアルゴリズムが含まれます
4 つのネットワーク構造は次のように比較されます。
ノードの検出
ノード検出は、ノードが P2P ネットワークにアクセスするための最初のステップです。ノード検出は、次の 2 つのタイプに分類できます。
- 初期ノード検出:
ノードは、一度も実行されていない真新しいノードであることを指します。このノードには、ネットワーク内の他のノードのデータがありません。この時点で、ノードは、ハードコーディングされたシード ノードのみに依存できることがわかります。ノードを使用してP2Pネットワークの情報を取得します。
- 既知のノードの検出:
ノードは以前から動作しており、ノード データベースにはネットワーク内の他のノードの情報が保存されていますが、このとき、ノードはノード データベースにまとめられたノードを利用して P2P ネットワークの情報を取得できることがわかりました。独自のネットワークトポロジを構築するため
シードノード"P2P ネットワークでは、最初のノードは起動時に長期的に安定したノードを介してネットワーク内の他のノードを迅速に検出します。"シードノード
DNS-Seed：
(一般に、シード ノード情報はコード内にハードコーディングされます。) 一般に、シード ノードは 2 つのタイプに分類できます。"としても知られている"DNSシードノード
IP-Seed：
, DNS は、インターネットによって提供されるドメイン名クエリ サービスです。ドメイン名と IP アドレスをマッピングし、分散データベースに保存します。DNS サーバーにアクセスするときは、ドメイン名を指定します。DNS サーバーは、返されたドメイン名 IP アドレス
つまり、シード ノードの IP アドレスはコードにハードコーディングされており、これらのハードコーディングされたノードのアドレスはシード ノードと呼ばれます。
KDA アルゴリズム
HashTable
Kademlia は分散ハッシュ テーブル (DHT) テクノロジーです。他の DHT テクノロジーと比較して、KDA アルゴリズムは XOR アルゴリズムを使用してノード間の距離を計算し、新しい DHT トポロジを確立します。このアルゴリズムにより、ルートのクエリ速度が大幅に向上します。 。
ハッシュ テーブルは、キーと値のペアを格納するためのコンテナです。キーと値のペアは、キー/値ペアと呼ばれます。ハッシュ テーブルのデータ構造には N 個のバケットが含まれています。特定のハッシュ テーブルの場合、N (バケットの数) は通常固定されています。各バケットに 0 ～ N-1 の番号を付けることができ、バケットはキーと値のペアを保存するために使用されます。これは、複数のキーと値のペアを保存する動的配列として単純に理解できます。
HashCollision
次の図は、ハッシュ テーブルの検索原理を示しています。キーを介して簡単かつ迅速に値を取得できます。特定のキーを使用して検索する場合、まず、特定のハッシュ関数を使用してキーのハッシュ値を計算し、取得した値を取得します。ハッシュ値は通常整数であり、ハッシュ値を使用して N (バケット数) (除算の余り) に対してモジュロ演算を実行し、対応するバケット番号を計算できます。
ハッシュ テーブルに関して言えば、ハッシュ テーブルの衝突について言及する必要があります。2 つの異なる Key が同じハッシュ値を取得するためにハッシュ計算を実行する場合、いわゆるハッシュ関数の衝突です。これが発生すると、2 つのキーと対応する 2 つのキーが衝突します。 -値のペアは同じバケット (バケット) に保存されます。もう 1 つのハッシュ衝突は、計算されたハッシュ値が異なるにもかかわらず、モジュロ演算後に同じバケット番号が得られることです。場合によっては、2 つのキーと値のペアが保存されますハッシュ衝突の原理を次の図に示します。
データを保存するときにハッシュ テーブルにまったく衝突がない場合、各バケットには 0 または 1 個のキーと値のペアしか存在しないため、検索が非常に高速になります。衝突により、特定のバケットに多数のキーと値のペアが格納されるため、キーを検索する際に、このバケット内のキーが同じかどうかを 1 つずつ比較する必要があり、検索効率が非常に低くなります~
分散ハッシュテーブル
分散ハッシュ テーブルは概念的には従来のハッシュ テーブルに似ていますが、違いは、従来のハッシュ テーブルは主に単一マシン上の特定のソフトウェアで使用され、分散ハッシュ テーブルは主に分散システムで使用されることです（現時点では、分散 (システムのノードは一般にハッシュ テーブル内のバケットとして理解できます)。分散ハッシュ テーブルは主に大量 (膨大なデータも) のデータを保存するために使用されます。分散ハッシュ テーブルの原理は次の図に示されています。 :
ソースコード分析
Ethereum の基盤となる P2PServer は、大きく次の 3 つの層に分けることができます。
- 上層：
イーサリアムにおける各プロトコルの具体的な実装
- 中間層:
Ethereum の p2p 通信リンク層は、監視の開始、新しい接続の処理、または接続の維持を担当します。
- 最下層：
イーサリアムのデータ通信ネットワークの IO 層は、主にルーティング テーブルの管理とデータベースの読み取りおよび書き込み操作を担当します。
テーブル構造
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L40
const (
alpha = 3 // Kademlia concurrency factor
bucketSize = 16 // Kademlia bucket size
maxReplacements = 10 // Size of per-bucket replacement list
// We keep buckets for the upper 1/15 of distances because
// it's very unlikely we'll ever encounter a node that's closer.
hashBits = len(common.Hash{}) * 8
nBuckets = hashBits / 15 // Number of buckets
bucketMinDistance = hashBits - nBuckets // Log distance of closest bucket
// IP address limits.
bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24
tableIPLimit, tableSubnet = 10, 24
refreshInterval = 30 * time.Minute
revalidateInterval = 10 * time.Second
copyNodesInterval = 30 * time.Second
seedMinTableTime = 5 * time.Minute
seedCount = 30
seedMaxAge = 5 * 24 * time.Hour
)
type Table struct {
mutex sync.Mutex // protects buckets, bucket content, nursery, rand
buckets [nBuckets]*bucket // index of known nodes by distance
nursery []*node // bootstrap nodes
rand *mrand.Rand // source of randomness, periodically reseeded
ips netutil.DistinctNetSet
log log.Logger
db *enode.DB // database of known nodes
net transport
refreshReq chan chan struct{}
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
nodeAddedHook func(*node) // for testing
}
type bucket struct {
entries []*node // live entries, sorted by time of last contact
replacements []*node // recently seen nodes to be used if revalidation fails
ips netutil.DistinctNetSet
}
テーブルのデータ構造は次のようになります。
いくつかの重要な変数:
バケット: K バケット。各 K バケットにはノード (最近のアクティビティに従って降順でソート) が含まれ、既知のノード インデックスを距離別にリストするために使用されます。
ナーサリー: シード ノード。ノードが開始されると、最大 35 個のシード ノードをリンクできます。そのうち 5 個はイーサリアムによって正式に指定され、残りの 30 個はデータベースから抽出されます。
db: P2P ノードを保存するために使用されるデータベース (イーサリアムには 2 つあり、もう 1 つはチェーン上のデータを保存するために使用されます)
- freshReq: K-bucket イベントを更新するパイプライン
テーブルの作成
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L102
func newTable(t transport, db *enode.DB, bootnodes []*enode.Node, log log.Logger) (*Table, error) {
tab := &Table{
net: t,
db: db,
refreshReq: make(chan chan struct{}),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
closed: make(chan struct{}),
rand: mrand.New(mrand.NewSource(0)),
ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit},
log: log,
}
if err := tab.setFallbackNodes(bootnodes); err != nil {
return nil, err
}
for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
tab.seedRand()
tab.loadSeedNodes()
return tab, nil
}
newTable 関数は、新しいテーブルを作成するために使用されます。
上記のコードでは、まず Table オブジェクトのタブが受信パラメータで初期化され、setFallbackNodes 関数が呼び出されて初期リンク ノードが設定されます (つまり、5 つのナース ノードを取得します。これは、後でテーブルが空であるか、またはテーブルが空である場合に使用されます)。データベースにノード情報がありません。ネットワークをリンクします)、関数 ValidateComplete と組み合わせた for ループを使用して、ノードが有効かどうかを確認します。
for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
次に、K バケットを初期化します。
tab.seedRand()
tab.loadSeedNodes()
return tab, nil
次に、table.buckets から 30 個のノードをランダムに選択し、シード ノードを対応するバケットにロードします。
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L302
func (tab *Table) loadSeedNodes() {
seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
seeds = append(seeds, tab.nursery...)
for i := range seeds {
seed := seeds[i]
age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
tab.addSeenNode(seed)
}
}
loadSeedNodes 関数の具体的な実装は次のとおりです (ここでの seedCount は、table.go の先頭で定義されているグローバル変数で、値は 30 です)。
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L458
func (tab *Table) addSeenNode(n *node) {
if n.ID() == tab.self().ID() {
return
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
if contains(b.entries, n.ID()) {
// Already in bucket, don't add.
return
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, n)
return
}
if !tab.addIP(b, n.IP()) {
// Can't add: IP limit reached.
return
}
// Add to end of bucket:
b.entries = append(b.entries, n)
b.replacements = deleteNode(b.replacements, n)
n.addedAt = time.Now()
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(n)
}
}
ここの addSeenNode は、バケットにノードを追加するために使用されます。ここでは、追加するノードが既に存在するかどうか、およびバケットがいっぱいかどうかを確認します。いっぱいの場合は、tab.addReplacement(b, n) を呼び出してノードを追加しますを置換リストに追加し、IP を追加してバケットを更新します。
イベントリスナー
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L55
refreshInterval = 30 * time.Minute
revalidateInterval = 10 * time.Second
copyNodesInterval = 30 * time.Second
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L218
// loop schedules runs of doRefresh, doRevalidate and copyLiveNodes.
func (tab *Table) loop() {
var (
revalidate = time.NewTimer(tab.nextRevalidateTime())
refresh = time.NewTicker(refreshInterval)
copyNodes = time.NewTicker(copyNodesInterval)
refreshDone = make(chan struct{}) // where doRefresh reports completion
revalidateDone chan struct{} // where doRevalidate reports completion
waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
)
defer refresh.Stop()
defer revalidate.Stop()
defer copyNodes.Stop()
// Start initial refresh.
go tab.doRefresh(refreshDone)
loop:
for {
select {
ループ関数は table.go のメイン ループです。後で使用するローカル変数は関数の先頭で定義され、バケットは deRefresh によって更新されます。ここでのループは、K バケットを 30 分ごとに自動的に更新します。 K バケット ノードに 10 秒ごとに ping できるかどうかを確認し、K バケット内のノードを安定したノードとして 30 秒ごとにローカル データベースに 5 分以上保存できるかどうかを確認します。
tab.seedRand()
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case <-refresh.C: //k バケット イベントを定期的に更新します。refreshInterval=30 分
waiting = append(waiting, req)
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case req := <-tab.refreshReq: //k バケットのリクエスト イベントを更新します
for _, ch := range waiting {
close(ch)
}
waiting, refreshDone = nil, nil
case <-refreshDone: // doRefresh が完了しました
revalidateDone = make(chan struct{})
go tab.doRevalidate(revalidateDone)
case <-revalidate.C: // k-bucket ノードの有効性を検証します (10 秒)
revalidate.Reset(tab.nextRevalidateTime())
revalidateDone = nil
case <-revalidateDone: // K バケット ノードの有効性の検証が完了しました
go tab.copyLiveNodes()
case <-tab.closeReq:
break loop
}
}
if refreshDone != nil {
<-refreshDone
}
for _, ch := range waiting {
close(ch)
}
if revalidateDone != nil {
<-revalidateDone
}
close(tab.closed)
case <-copyNodes.C: // 定期的にノードをデータベースに保存します (30 秒)。ノードが k-bucket に 5 分以上存在する場合、そのノードは安定したノードとみなされます
ノード検索
// getNode returns the node with the given ID or nil if it isn't in the table.
func (tab *Table) getNode(id enode.ID) *enode.Node {
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(id)
for _, e := range b.entries {
if e.ID() == id {
return unwrapNode(e)
}
}
return nil
}
getNode 関数は ID でノードを検索するために使用され、ノードが存在しない場合は nil を返します。
ノードの検出
// filedir:go-ethereum-1.10.2\p2p\discover\table.go L278
func (tab *Table) doRefresh(done chan struct{}) {
defer close(done)
tab.loadSeedNodes()
// Run self lookup to discover new neighbor nodes.
tab.net.lookupSelf()
for i := 0; i < 3; i++ {
tab.net.lookupRandom()
}
}
イーサリアム分散ネットワークは構造化ネットワーク モデルを採用しており, その実装スキームは Kademlia プロトコルを使用します. 以下にノード検出を簡単に紹介します. イーサリアムでは, k の値は 16, つまり、各 k バケットには 16 ノードが含まれます. 256 k-buckets , K-buckets はノードの NodeId、Distance、Endpoint、IP およびその他の情報を記録し、ターゲット ノードからの距離に従って並べ替えます。ノードの検索は doRefresh() によって実装されます。
// filedir: go-ethereum-1.10.2\p2p\discover\table.go L302
func (tab *Table) loadSeedNodes() {
seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
seeds = append(seeds, tab.nursery...)
for i := range seeds {
seed := seeds[i]
age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
tab.addSeenNode(seed)
}
}
// filedir:go-ethereum-1.10.2\p2p\enode\nodedb.go L440
// QuerySeeds retrieves random nodes to be used as potential seed nodes
// for bootstrapping.
func (db *DB) QuerySeeds(n int, maxAge time.Duration) []*Node {
var (
now = time.Now()
nodes = make([]*Node, 0, n)
it = db.lvl.NewIterator(nil, nil)
id ID
)
defer it.Release()
seek:
for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ {
ctr := id[0]
rand.Read(id[:])
id[0] = ctr + id[0]%16
it.Seek(nodeKey(id))
n := nextNode(it)
if n == nil {
id[0] = 0
continue seek // iterator exhausted
}
if now.Sub(db.LastPongReceived(n.ID(), n.IP())) > maxAge {
continue seek
}
for i := range nodes {
if nodes[i].ID() == n.ID() {
continue seek // duplicate
}
}
nodes = append(nodes, n)
}
return nodes
}
上記のコードからわかるように、まず tab.loadSeedNodes() を呼び出してデータベースからノードをロードし、テーブルに挿入します。
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L281
// lookupSelf looks up our own node ID.
// This is needed to satisfy the transport interface.
func (t *UDPv5) lookupSelf() []*enode.Node {
return t.newLookup(t.closeCtx, t.Self().ID()).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L292
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
return t.lookupWorker(n, target)
})
}
func newLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *lookup {
it := &lookup{
tab: tab,
queryfunc: q,
asked: make(map[enode.ID]bool),
seen: make(map[enode.ID]bool),
result: nodesByDistance{target: target},
replyCh: make(chan []*node, alpha),
cancelCh: ctx.Done(),
queries: -1,
}
// Don't query further if we hit ourself.
// Unlikely to happen often in practice.
it.asked[tab.self().ID()] = true
return it
}
次に、lookupSelf を使用して新しいノードを検出します。ここでは、最初に現在のノードの ID を使用して newLookup を実行し、隣接ノードを検出します。
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L275
func (t *UDPv5) lookupRandom() []*enode.Node {
return t.newRandomLookup(t.closeCtx).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L287
func (t *UDPv5) newRandomLookup(ctx context.Context) *lookup {
var target enode.ID
crand.Read(target[:])
return t.newLookup(ctx, target)
}
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
return t.lookupWorker(n, target)
})
}
// lookupWorker performs FINDNODE calls against a single node during lookup.
func (t *UDPv5) lookupWorker(destNode *node, target enode.ID) ([]*node, error) {
var (
dists = lookupDistances(target, destNode.ID())
nodes = nodesByDistance{target: target}
err error
)
var r []*enode.Node
r, err = t.findnode(unwrapNode(destNode), dists)
if err == errClosed {
return nil, err
}
for _, n := range r {
if n.ID() != t.Self().ID() {
nodes.push(wrapNode(n), findnodeResultLimit)
}
}
return nodes.entries, err
}
// filedir:go-ethereum-1.10.2\p2p\discover\v5_udp.go L362
// findnode calls FINDNODE on a node and waits for responses.
func (t *UDPv5) findnode(n *enode.Node, distances []uint) ([]*enode.Node, error) {
resp := t.call(n, v5wire.NodesMsg, &v5wire.Findnode{Distances: distances})
return t.waitForNodes(resp, distances)
}
func (t *UDPv5) call(node *enode.Node, responseType byte, packet v5wire.Packet) *callV5 {
c := &callV5{
node: node,
packet: packet,
responseType: responseType,
reqid: make([]byte, 8),
ch: make(chan v5wire.Packet, 1),
err: make(chan error, 1),
}
// Assign request ID.
crand.Read(c.reqid)
packet.SetRequestID(c.reqid)
// Send call to dispatch.
select {
case t.callCh <- c:
case <-t.closeCtx.Done():
c.err <- errClosed
}
return c
}
最後に、ランダムなターゲットが検索のために実行されます。
サービス体系
// filedir:go-ethereum-1.10.2\p2p\server.go L160
// Server manages all peer connections.
type Server struct {
// Config fields may not be modified while the server is running.
Config
// Hooks for testing. These are useful because we can inhibit
// the whole protocol stack.
newTransport func(net.Conn, *ecdsa.PublicKey) transport
newPeerHook func(*Peer)
listenFunc func(network, addr string) (net.Listener, error)
lock sync.Mutex // protects running
running bool
listener net.Listener
ourHandshake *protoHandshake
loopWG sync.WaitGroup // loop, listenLoop
peerFeed event.Feed
log log.Logger
nodedb *enode.DB
localnode *enode.LocalNode
ntab *discover.UDPv4
DiscV5 *discover.UDPv5
discmix *enode.FairMix
dialsched *dialScheduler
// Channels into the run loop.
quit chan struct{}
addtrusted chan *enode.Node
removetrusted chan *enode.Node
peerOp chan peerOpFunc
peerOpDone chan struct{}
delpeer chan peerDrop
checkpointPostHandshake chan *conn
checkpointAddPeer chan *conn
// State of run loop and listenLoop.
inboundHistory expHeap
}
サーバー側のデータ構造は次のとおりです。
// Config holds Server options.
type Config struct {
PrivateKey *ecdsa.PrivateKey `toml:"-"`
MaxPeers int
MaxPendingPeers int `toml:",omitempty"`
DialRatio int `toml:",omitempty"`
NoDiscovery bool
DiscoveryV5 bool `toml:",omitempty"`
Name string `toml:"-"`
BootstrapNodes []*enode.Node
BootstrapNodesV5 []*enode.Node `toml:",omitempty"`
StaticNodes []*enode.Node
TrustedNodes []*enode.Node
NetRestrict *netutil.Netlist `toml:",omitempty"`
NodeDatabase string `toml:",omitempty"`
Protocols []Protocol `toml:"-"`
ListenAddr string
NAT nat.Interface `toml:",omitempty"`
Dialer NodeDialer `toml:"-"`
NoDial bool `toml:",omitempty"`
EnableMsgEvents bool
Logger log.Logger `toml:",omitempty"`
clock mclock.Clock
}
#####
サーバー構成 (ローカル ノードの秘密キー、ダイヤルアップ比率、ノード接続の最大数、ダイヤルアップ比率、イベント レコードなど):
新しいノード
// filedir:go-ethereum-1.10.2\p2p\server.go L318
func (srv *Server) AddPeer(node *enode.Node) {
srv.dialsched.addStatic(node)
}
// filedir:go-ethereum-1.10.2\p2p\dial.go L190
// addStatic adds a static dial candidate.
func (d *dialScheduler) addStatic(n *enode.Node) {
select {
case d.addStaticCh <- n:
case <-d.ctx.Done():
}
}
AddPeer 関数は特定のノードを追加するために使用され、その実装コードは次のとおりです。
// AddTrustedPeer adds the given node to a reserved whitelist which allows the
// node to always connect, even if the slot are full.
func (srv *Server) AddTrustedPeer(node *enode.Node) {
select {
case srv.addtrusted <- node:
case <-srv.quit:
}
}
AddTrustedPeer 関数は、信頼できるノードを追加するために使用されます。
ノードを削除する
// filedir:go-ethereum-1.10.2\p2p\server.go L325
func (srv *Server) RemovePeer(node *enode.Node) {
var (
ch chan *PeerEvent
sub event.Subscription
)
// Disconnect the peer on the main loop.
srv.doPeerOp(func(peers map[enode.ID]*Peer) {
srv.dialsched.removeStatic(node)
if peer := peers[node.ID()]; peer != nil {
ch = make(chan *PeerEvent, 1)
sub = srv.peerFeed.Subscribe(ch)
peer.Disconnect(DiscRequested)
}
})
// Wait for the peer connection to end.
if ch != nil {
defer sub.Unsubscribe()
for ev := range ch {
if ev.Peer == node.ID() && ev.Type == PeerEventTypeDrop {
return
}
}
}
}
// filedir:go-ethereum-1.10.2\p2p\dial.go L198
// removeStatic removes a static dial candidate.
func (d *dialScheduler) removeStatic(n *enode.Node) {
select {
case d.remStaticCh <- n:
case <-d.ctx.Done():
}
}
RemovePeer 関数は、ノードを削除し、ノードから切断するために使用されます。
// RemoveTrustedPeer removes the given node from the trusted peer set.
func (srv *Server) RemoveTrustedPeer(node *enode.Node) {
select {
case srv.removetrusted <- node:
case <-srv.quit:
}
}
RemoveTrustedPeer 関数は、信頼できるピアを削除するために使用されます。
サービスを終了する
func (srv *Server) Stop() {
srv.lock.Lock()
if !srv.running {
srv.lock.Unlock()
return
}
srv.running = false
if srv.listener != nil {
// this unblocks listener Accept
srv.listener.Close()
}
close(srv.quit)
srv.lock.Unlock()
srv.loopWG.Wait()
}
Stop 関数はノード操作を終了するために使用されます。具体的なコードは次のとおりです。
サービス開始
// filedir:go-ethereum-1.10.2\p2p\server.go L433
func (srv *Server) Start() (err error) {
srv.lock.Lock()
defer srv.lock.Unlock()
if srv.running {
return errors.New("server already running")
}
srv.running = true
srv.log = srv.Config.Logger
if srv.log == nil {
srv.log = log.Root()
}
if srv.clock == nil {
srv.clock = mclock.System{}
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
// static fields
if srv.PrivateKey == nil {
return errors.New("Server.PrivateKey must be set to a non-nil key")
}
if srv.newTransport == nil {
srv.newTransport = newRLPX
}
if srv.listenFunc == nil {
srv.listenFunc = net.Listen
}
srv.quit = make(chan struct{})
srv.delpeer = make(chan peerDrop)
srv.checkpointPostHandshake = make(chan *conn)
srv.checkpointAddPeer = make(chan *conn)
srv.addtrusted = make(chan *enode.Node)
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
if err := srv.setupLocalNode(); err != nil {
return err
}
if srv.ListenAddr != "" {
if err := srv.setupListening(); err != nil {
return err
}
}
if err := srv.setupDiscovery(); err != nil {
return err
}
srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}
go-ethereum-1.10.2\p2p\server.go にある start 関数は、P2P ノードを起動するために使用されます。
// Start starts running the server.
// Servers can not be re-used after stopping.
func (srv *Server) Start() (err error) {
srv.lock.Lock()
defer srv.lock.Unlock()
if srv.running {
return errors.New("server already running")
}
srv.running = true
srv.log = srv.Config.Logger
if srv.log == nil {
srv.log = log.Root()
}
if srv.clock == nil {
srv.clock = mclock.System{}
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
// static fields
if srv.PrivateKey == nil {
return errors.New("Server.PrivateKey must be set to a non-nil key")
}
if srv.newTransport == nil {
srv.newTransport = newRLPX
}
if srv.listenFunc == nil {
srv.listenFunc = net.Listen
}
srv.quit = make(chan struct{})
srv.delpeer = make(chan peerDrop)
srv.checkpointPostHandshake = make(chan *conn)
srv.checkpointAddPeer = make(chan *conn)
srv.addtrusted = make(chan *enode.Node)
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
ここでは、まず現在のノードが実行状態かどうかを確認し、実行中の場合は直接戻ってエラー メッセージを表示します。そうでない場合は、srv.running を true に設定し、サービス起動プロセスに入り、ログが有効かどうかを確認します。などを実行し、構成 P2P サービス情報を初期化します。
次に、setupLocalNode を呼び出してローカル ノードを起動し、ローカル モニターを確立してから、DiscoveryV5 ネットワーク プロトコルを構成してノード ルーティング テーブルを生成します。
srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}
次に、setupDialScheduler を呼び出してアクティブなダイヤルアップ接続プロセスを開始し、ピアを維持するコルーチンを開きます。
func (srv *Server) setupDialScheduler() {
config := dialConfig{
self: srv.localnode.ID(),
maxDialPeers: srv.maxDialedConns(),
maxActiveDials: srv.MaxPendingPeers,
log: srv.Logger,
netRestrict: srv.NetRestrict,
dialer: srv.Dialer,
clock: srv.clock,
}
if srv.ntab != nil {
config.resolver = srv.ntab
}
if config.dialer == nil {
config.dialer = tcpDialer{&net.Dialer{Timeout: defaultDialTimeout}}
}
srv.dialsched = newDialScheduler(config, srv.discmix, srv.SetupConn)
for _, n := range srv.StaticNodes {
srv.dialsched.addStatic(n)
}
}
setupDialScheduler コードは次のとおりです。ここでは、newDialScheduler を通じて接続が確立されます。パラメータ discmix は、接続をアクティブに確立するときに設定されるノードを決定します。これは反復子であり、同時に setupConn 接続確立関数が渡されます。
// filedir:go-ethereum-1.10.2\p2p\dial.go L162
func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupFunc) *dialScheduler {
d := &dialScheduler{
dialConfig: config.withDefaults(),
setupFunc: setupFunc,
dialing: make(map[enode.ID]*dialTask),
static: make(map[enode.ID]*dialTask),
peers: make(map[enode.ID]connFlag),
doneCh: make(chan *dialTask),
nodesIn: make(chan *enode.Node),
addStaticCh: make(chan *enode.Node),
remStaticCh: make(chan *enode.Node),
addPeerCh: make(chan *conn),
remPeerCh: make(chan *conn),
}
d.lastStatsLog = d.clock.Now()
d.ctx, d.cancel = context.WithCancel(context.Background())
d.wg.Add(2)
go d.readNodes(it)
go d.loop(it)
return d
}
newDialScheduler 関数は次のとおりです。ここで、ノードは d.readNodes(it) を通じてイテレータから取得され、接続用のチャネルを通じて d.loop(it) に渡されます。
サービス監視
func (srv *Server) setupListening() error {
// Launch the listener.
listener, err := srv.listenFunc("tcp", srv.ListenAddr)
if err != nil {
return err
}
srv.listener = listener
srv.ListenAddr = listener.Addr().String()
// Update the local node record and map the TCP listening port if NAT is configured.
if tcp, ok := listener.Addr().(*net.TCPAddr); ok {
srv.localnode.Set(enr.TCP(tcp.Port))
if !tcp.IP.IsLoopback() && srv.NAT != nil {
srv.loopWG.Add(1)
go func() {
nat.Map(srv.NAT, srv.quit, "tcp", tcp.Port, tcp.Port, "ethereum p2p")
srv.loopWG.Done()
}()
}
}
srv.loopWG.Add(1)
go srv.listenLoop()
return nil
}
上記のサービス起動プロセスには、イベントをリッスンするために使用される setupListening 関数があり、具体的なコードは次のとおりです。
// listenLoop runs in its own goroutine and accepts
// inbound connections.
func (srv *Server) listenLoop() {
srv.log.Debug("TCP listener up", "addr", srv.listener.Addr())
// The slots channel limits accepts of new connections.
tokens := defaultMaxPendingPeers
if srv.MaxPendingPeers > 0 {
tokens = srv.MaxPendingPeers
}
slots := make(chan struct{}, tokens)
for i := 0; i < tokens; i++ {
slots <- struct{}{}
}
// Wait for slots to be returned on exit. This ensures all connection goroutines
// are down before listenLoop returns.
defer srv.loopWG.Done()
defer func() {
for i := 0; i < cap(slots); i++ {
<-slots
}
}()
for {
// Wait for a free slot before accepting.
<-slots
var (
fd net.Conn
err error
lastLog time.Time
)
for {
fd, err = srv.listener.Accept()
if netutil.IsTemporaryError(err) {
if time.Since(lastLog) > 1*time.Second {
srv.log.Debug("Temporary read error", "err", err)
lastLog = time.Now()
}
time.Sleep(time.Millisecond * 200)
continue
} else if err != nil {
srv.log.Debug("Read error", "err", err)
slots <- struct{}{}
return
}
break
}
remoteIP := netutil.AddrIP(fd.RemoteAddr())
if err := srv.checkInboundConn(remoteIP); err != nil {
srv.log.Debug("Rejected inbound connection", "addr", fd.RemoteAddr(), "err", err)
fd.Close()
slots <- struct{}{}
continue
}
if remoteIP != nil {
var addr *net.TCPAddr
if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok {
addr = tcp
}
fd = newMeteredConn(fd, true, addr)
srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr())
}
go func() {
srv.SetupConn(fd, inboundConn, nil)
slots <- struct{}{}
}()
}
}
上記のコードでは、別の srv.listenLoop() が呼び出されます。これは、ポートをリッスンして外部リクエストを受信する無限ループのゴルーチンです。
// SetupConn runs the handshakes and attempts to add the connection
// as a peer. It returns when the connection has been added as a peer
// or the handshakes have failed.
func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) error {
c := &conn{fd: fd, flags: flags, cont: make(chan error)}
if dialDest == nil {
c.transport = srv.newTransport(fd, nil)
} else {
c.transport = srv.newTransport(fd, dialDest.Pubkey())
}
err := srv.setupConn(c, flags, dialDest)
if err != nil {
c.close(err)
}
return err
}
ここの SetupConn は主にハンドシェイク プロトコルを実行し、ピア オブジェクトとしてリンクを作成しようとします。
func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) error {
// Prevent leftover pending conns from entering the handshake.
srv.lock.Lock()
running := srv.running
srv.lock.Unlock()
if !running {
return errServerStopped
}
// If dialing, figure out the remote public key.
var dialPubkey *ecdsa.PublicKey
上記のコードでは、ハンドシェイク プロトコルを実行するために使用される srv.setupConn(c, flags, digialDest) 関数が再度呼び出されます。
dialPubkey = new(ecdsa.PublicKey)
if err := dialDest.Load((*enode.Secp256k1)(dialPubkey)); err != nil {
err = errors.New("dial destination doesn't have a secp256k1 public key")
srv.log.Trace("Setting up connection failed", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
return err
}
}
// Run the RLPx handshake.
remotePubkey,if digialDest != nil { // dest=nil パッシブ接続、dest != nil アクティブ接続
if err != nil {
srv.log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
return err
}
if dialDest != nil {
c.node = dialDest
} else {
c.node = nodeFromConn(remotePubkey, c.fd)
}
clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags)
err = srv.checkpoint(c, srv.checkpointPostHandshake)
if err != nil {
clog.Trace("Rejected peer", "err", err)
return err
}
// Run the capability negotiation handshake.
phs,err := c.doEncHandshake(srv.PrivateKey) // 公開鍵交換、共有秘密鍵の RLPx レベルでのハンドシェイクを決定します
if err != nil {
clog.Trace("Failed p2p handshake", "err", err)
return err
}
if id := c.node.ID(); !bytes.Equal(crypto.Keccak256(phs.ID), id[:]) {
clog.Trace("Wrong devp2p handshake identity", "phsid", hex.EncodeToString(phs.ID))
return DiscUnexpectedIdentity
}
c.caps, c.name = phs.Caps, phs.Name
err = srv.checkpoint(c,err := c.doProtoHandshake(srv.ourHandshake) // プロトコル レベルでのハンドシェイク、つまり p2p ハンドシェイク、行ったり来たり
if err != nil {
clog.Trace("Rejected peer", "err", err)
return err
}
return nil
}
srv.checkpointAddPeer) // ステータスの検証
// filedir:go-ethereum-1.10.2\p2p\transport.go L123
func (t *rlpxTransport) doEncHandshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
t.conn.SetDeadline(time.Now().Add(handshakeTimeout))
return t.conn.Handshake(prv)
}
キー ハンドシェイクは deEncHandshake 関数を通じて実装され、関数内で Handshake() 関数が呼び出されます。
// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go L253
func (c *Conn) Handshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
var (
sec Secrets
err error
)
ハンドシェイク コードは次のとおりです。アクティブ ハンドシェイクかパッシブ ハンドシェイクかに応じて、対応するハンドシェイク ロジックが実行されます。
sec, err = initiatorEncHandshake(c.conn, prv,if c.dialDest != nil { //アクティブなハンドシェイク
c.dialDest) //鍵検証ハンドシェイクを積極的に開始し、共有鍵を確認します
sec, err = receiverEncHandshake(c.conn, prv)
}
if err != nil {
return nil, err
}
c.InitWithSecrets(sec)
return sec.remote, err
}
} else { // パッシブハンドシェイク
// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go L477
func initiatorEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey, remote *ecdsa.PublicKey) (s Secrets, err error) {
h := &encHandshake{initiator: true, remote: ecies.ImportECDSAPublic(remote)}
authMsg, err := h.makeAuthMsg(prv)
if err != nil {
return s, err
}
authPacket, err := sealEIP8(authMsg, h)
if err != nil {
return s, err
}
if _, err = conn.Write(authPacket); err != nil {
return s, err
}
authRespMsg := new(authRespV4)
authRespPacket, err := readHandshakeMsg(authRespMsg, encAuthRespLen, prv, conn)
if err != nil {
return s, err
}
if err := h.handleAuthResp(authRespMsg); err != nil {
return s, err
}
return h.secrets(authPacket, authRespPacket)
}
ハンドシェイクをアクティブに開始するプロセスは次のとおりです。ここでは、makeAuthMsg が呼び出されて、署名、公開キー、ランダムなノンスによって生成された署名に対応するバージョン番号を含む認証 ID 情報が生成され、その後、sealEIP8 メソッドが実行されます。 rlpx エンコードを実行するために呼び出され、暗号化されたハンドシェイクが開始され、受信されます。 authResp メッセージを返し、検証して復号し、相手の公開キーを取得して、AES、MAC を生成します。
func receiverEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey) (s Secrets, err error) {
authMsg := new(authMsgV4)
authPacket, err := readHandshakeMsg(authMsg, encAuthMsgLen, prv, conn)
if err != nil {
return s, err
}
h := new(encHandshake)
if err := h.handleAuthMsg(authMsg, prv); err != nil {
return s, err
}
authRespMsg, err := h.makeAuthResp()
if err != nil {
return s, err
}
var authRespPacket []byte
if authMsg.gotPlain {
authRespPacket, err = authRespMsg.sealPlain(h)
} else {
authRespPacket, err = sealEIP8(authRespMsg, h)
}
if err != nil {
return s, err
}
if _, err = conn.Write(authRespPacket); err != nil {
return s, err
}
return h.secrets(authPacket, authRespPacket)
}
receiverEncHandshake は次のとおりであり、initiatorEncHandshake とほぼ同じです。
func (t *rlpxTransport) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err error) {
werr := make(chan error, 1)
go func() { werr <- Send(t, handshakeMsg, our) }()
if their, err = readProtocolHandshake(t); err != nil {
<-werr // make sure the write terminates too
return nil, err
}
if err := <-werr; err != nil {
return nil, fmt.Errorf("write error: %v", err)
}
// If the protocol version supports Snappy encoding, upgrade immediately
t.conn.SetSnappy(their.Version >= snappyProtocolVersion)
return their, nil
}
次に、doProtoHandshake を通じてプロトコル ハンドシェイク操作を完了し、ここで send を呼び出してハンドシェイク操作を送信します。次に、readProtocolHandshake を通じて返された情報を読み取り、以下を確認します。
サービスサイクル
func (srv *Server) run() {
srv.log.Info("Started P2P networking", "self", srv.localnode.Node().URLv4())
defer srv.loopWG.Done()
defer srv.nodedb.Close()
defer srv.discmix.Close()
defer srv.dialsched.stop()
var (
peers = make(map[enode.ID]*Peer)
inboundCount = 0
trusted = make(map[enode.ID]bool, len(srv.TrustedNodes))
)
for _, n := range srv.TrustedNodes {
trusted[n.ID()] = true
}
running:
for {
select {
case <-srv.quit:
break running
case n := <-srv.addtrusted:
srv.log.Trace("Adding trusted node", "node", n)
trusted[n.ID()] = true
if p, ok := peers[n.ID()]; ok {
p.rw.set(trustedConn, true)
}
case n := <-srv.removetrusted:
srv.log.Trace("Removing trusted node", "node", n)
delete(trusted, n.ID())
if p, ok := peers[n.ID()]; ok {
p.rw.set(trustedConn, false)
}
case op := <-srv.peerOp:
op(peers)
srv.peerOpDone <- struct{}{}
case c := <-srv.checkpointPostHandshake:
if trusted[c.node.ID()] {
c.flags |= trustedConn
}
c.cont <- srv.postHandshakeChecks(peers, inboundCount, c)
case c := <-srv.checkpointAddPeer:
err := srv.addPeerChecks(peers, inboundCount, c)
if err == nil {
p := srv.launchPeer(c)
peers[c.node.ID()] = p
srv.log.Debug("Adding p2p peer", "peercount", len(peers), "id", p.ID(), "conn", c.flags, "addr", p.RemoteAddr(), "name", p.Name())
srv.dialsched.peerAdded(c)
if p.Inbound() {
inboundCount++
}
}
c.cont <- err
case pd := <-srv.delpeer:
// A peer disconnected.
d := common.PrettyDuration(mclock.Now() - pd.created)
delete(peers, pd.ID())
srv.log.Debug("Removing p2p peer", "peercount", len(peers), "id", pd.ID(), "duration", d, "req", pd.requested, "err", pd.err)
srv.dialsched.peerRemoved(pd.rw)
if pd.Inbound() {
inboundCount--
}
}
}
srv.log.Trace("P2P networking is spinning down")
if srv.ntab != nil {
srv.ntab.Close()
}
if srv.DiscV5 != nil {
srv.DiscV5.Close()
}
for _, p := range peers {
p.Disconnect(DiscQuitting)
}
for len(peers) > 0 {
p := <-srv.delpeer
p.log.Trace("<-delpeer (spindown)")
delete(peers, p.ID())
}
}
run 関数はサービスのメイン ループであり、サーバーの終了の監視、信頼ノードの追加、信頼ノードの削除、チェック ノードの追加などを行います。
ノード情報
// NodeInfo gathers and returns a collection of metadata known about the host.
func (srv *Server) NodeInfo() *NodeInfo {
node := srv.Self()
info := &NodeInfo{
Name: srv.Name,
Enode: node.URLv4(),
ID: node.ID().String(),
IP: &nb
