從代碼層深入理解以太坊P2P 網絡設計
前言
前言
本篇文章將對P2P 網絡發展進行簡單概述，同時將從源碼角度對以太坊中的節點發現機制、分佈式哈希表、節點查找、節點新增、節點移除等進行簡單介紹，並對其P2P 網絡安全性設計進行簡要分析。
基礎知識
基礎知識
基礎知識
基礎知識
P2P 網絡不同於傳統的CS 結構，在P2P 網絡中每個節點既可以是客戶端也可以是服務端，節點之間的通信協議一般直接通過Socket 實現。
一.集中式：
P2P 技術發展至今經歷了以下四個發展階段：
一.集中式：
是P2P 網絡模式中最簡單的路由方式，即存在一個中心節點，它保存了其他所有節點的索引信息，索引信息一般包括節點IP、端口、節點資源等，集中式路由的優點是結構簡單、實現容易，但缺點也很明顯，由於中心節點需要存儲所有節點的路由信息，當節點規模不斷擴展時，就很容易出現性能瓶頸，而且也存在單節點故障問題"二.分佈式："是指移除了中心節點，在P2P 節點之間建立隨機網絡，在新加入節點與P2P 網絡中的某個節點之間隨機建立連接通道，從而形成一個隨機拓撲結構，新節點加入該網絡時隨機選擇一個已經存在的節點並建立鄰居關係，在節點與鄰居節點建立連接後，還需要進行全網廣播，讓整個網絡知道該節點的存在
具體的廣播步驟是：該節點首先向鄰居節點廣播，鄰居節點收到廣播消息後，在繼續向自己的鄰居節點廣播，以此類推，這種廣播方式也被稱之為
，而泛洪機制的問題在於可控性差，其主要包括兩個較大的問題：一個是容易形成泛洪循環，比如節點A 給發出的消息結果節點B 到節點C，節點C 再廣播到節點A，這就形成了一個循環，另一個問題是消息響應分包，比如節點A 想要請求的資源被很多節點所擁有，那麼在短時間內，會出現大量節點同時向A 節點發送響應消息，這就很可能讓節點A瞬間奔潰
三.混合式：
而消除泛洪循環的方法可以藉鑑IP 網絡路由協議中有關泛洪廣播的控制，一種方法是對每一個查詢消息設置TTL值，泛洪消息每被轉發一次，TTL 值減1，當節點接受的TTL 為0 時，不再轉發消息，這樣可以避免查詢消息在網絡中產生死循環，還可以為泛洪消息設置唯一的標誌，對接收到的重複消息不再進行轉發從而規避死循環，解決響應風暴的方法可以在數據鏈路層進行網絡分段，減少消息跨段廣播
三.混合式：
混合式其實就是混合集中式和分佈式結構，網絡中存在很多超級節點組成的分佈式網絡，而每個超級節點有多個普通節點與它組成局部集中網絡，一個新的普通節點加入是可以先選擇一個超級節點進行通信，該超級節點再推送其他超級節點列表給新加入節點，加入節點根據列表中的超級節點狀態決定選擇那個具體的超級節點作為父節點，這種結構的泛洪廣播只是發生在超級節點之間，因此可以避免大規模泛洪問題，在實際應用中，混合式結構是相對靈活且比較有效的組網架構，實現難度也相對較小，因此目前較多系統基於混合式結構進行開發實現
四.結構化：
結構化P2P 網絡是一種分佈式網絡結構，與上面所講的分佈式結構不同，分佈式網絡就是一個隨機網絡，而結構化網絡則將所有節點按照某種結構進行有序組織，比如形成一個環狀網絡或樹狀網絡，結構化網絡在具體實現上普遍基於分佈式哈希表(Distributed Hash Table，DHI) 算法，具體的實現方案有Chord、Pasty、CAN、Kademlia 等算法
四種網絡結構對比如下：
節點發現
節點發現是任何節點接入P2P 網絡的第一步，節點發現可以分為兩種：
-初始節點發現：
指節點是一個全新的、從未運行的節點，該節點沒有網絡中的其他節點的任何數據，此時節點發現只能依靠節點中的硬編碼的種子節點獲得P2P 網絡的信息
-已知節點發現："節點之前運行過，節點數據庫中保存著網絡中的其他節點信息，此時節點發現可以依靠節點數據庫匯總的節點獲取P2P 網絡的信息，從而構建自己的網絡拓撲"種子節點
DNS-Seed：
在P2P 網絡中，初始節點在啟動時會通過一些長期穩定運行的節點快速發現網絡中的其他節點，這些節點被稱為"種子節點"(一般代碼中會硬編碼種子節點信息)，一般情況下種子節點可以分為兩種：
IP-Seed：
也被稱之為
，DNS 是互聯網提供的一種域名查詢服務，它將域名和IP 地址相互映射保存在一個分佈式的數據庫中，當我們訪問DNS 服務器時，給它提供一個域名，DNS 服務器會將該域名對應的IP地址返回
KDA算法
HashTable
即將種子節點的IP 地址硬編碼到代碼中去，硬編碼的這些節點的地址被稱為種子節點
KDA算法
HashCollision
Kademlia 是一種分佈式哈希表(DHT)技術，與其他DHT 技術相比，KDA 算法使用異或算法計算節點之間的距離，進而建立了全新的DHT 拓撲結構，這種算法可以極大地提高路由的查詢速度。
哈希表是用於存儲鍵值對的一種容器，鍵值對有被稱為Key/Value 對，哈希表數據結構中包含N 個bucket(桶)，對於某個具體的哈希表， N (桶的數量)通常是固定不變的，於是可以對每個桶編號，0~N-1，桶是用來存儲鍵值對的，可以簡單的將其理解為一個動態數組，裡面存放多個鍵值對。
下圖展示了哈希表的查找原理，我們可以方便快速地通過Key 來獲取value，當使用某個key 進行查找時，先用某個哈希函數計算這個key 的哈希值，得到的哈希值通常是一個整數，之後使用哈希值對N(桶數)進行取模運算(除法求餘數)，就可以算出對應的桶編號
說到哈希表不得不提一下哈希表碰撞，當兩個不同的Key 進行哈希計算得到相同的哈希值時，就是所謂的哈希函數碰撞，一旦出現這種情況，這兩個key對應的兩個鍵值對就會被存在在同一個桶中(bucket) 中，另一中散列碰撞是雖然計算出來的哈希值不同，但經過取模運算之後得到相同的桶編號，這時候也會將兩個鍵值對存儲在一個桶中，哈希碰撞原理如下圖所示：
如果某個哈希表在存儲數據時完全沒有碰撞，那麼每個桶裡都只有0 個或1 個鍵值對，這樣查找起來就非常快，反之，如果某個哈希表在存儲數據時出現嚴重碰撞，那麼就會導致某些桶裡存儲了很多鍵值對，那麼在查找key 的時候需要在這個桶裡面逐一對比key 是否相同，查找效率會變得很低~
分佈式哈希表
分佈式哈希表在概念上類似於傳統的哈希表，差異在於傳統的哈希表主要用於單機上的某個軟件中，分佈式哈希表主要用於分佈式系統(此時，分佈式系統的節點可以通俗的理解為hash 表中的bucket)，分佈式哈希表主要用於存儲大量(甚至海量)的數據，分佈式哈希表的原理如下圖所示：
源碼分析
以太坊底層的P2PServer 大致可以分為以下三層：
-頂層：
以太坊中各個協議的具體實現
-中層：
以太坊中的p2p 通信鏈路層，負責啟動監聽、處理新加入連接或維護連接
-底層：
// 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
}
以太坊中的數據通信網絡IO 層，主要負責路由表的管理以及數據庫的讀寫操作
表的結構
表數據結構如下所示：
關鍵的幾個變量：
nursery：種子節點，一個節點啟動的時候最多能夠鏈接35個種子節點，其中有五個是以太坊官方指定的，另外30個從數據庫裡面提取
- refreshReq：刷新K-桶事件的管道
db：用於存儲P2P節點的數據庫(以太坊中有兩個，另一個用於存儲鏈上數據)
// 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
}
refreshReq：刷新K-桶事件的管道
表的創建
for i := range tab.buckets {
tab.buckets[i] = &bucket{
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
newTable函數用於創建新的表：
tab.seedRand()
tab.loadSeedNodes()
return tab, nil
在上述代碼中首先使用傳入的參數初始化了一個Table的對象tab，調用setFallbackNodes函數設置初始鏈接節點(即獲得5個nursey節點，後面如果table為空或者數據庫中沒有節點信息時這些節點將被用於去鏈接網絡)，之後通過一個for循環結合函數ValidateComplete來驗證節點是否有效
// 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)
}
}
之後初始化K桶：
// 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)
}
}
之後從table.buckets中隨機取30個節點加載種子節點到相應的bucket:
loadSeedNodes函數的具體實現如下所示(這裡的seedCount為table.go中最上方定義的全局變量，值為30)：
// 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 {
這裡的addSeenNode即用於添加節點到bucket，在這裡會檢查要添加的節點是否已經存在以及bucket是否已滿，如果已滿則調用tab.addReplacement(b, n)將節點添加到replacement列表中去，之後添加IP，之後更新bucket：
tab.seedRand()
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
事件監聽
waiting = append(waiting, req)
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
loop函數是table.go中的主循環，在函數開頭出定義了後續會使用到的局部變量，之後通過deRefresh進行刷新桶操作，在這裡的loop循環會每隔30分鐘自動刷新一次K桶，每隔10秒鐘驗證K桶節點是否可以ping通，每30秒將K桶中存在超過5分鐘的節點存儲本地數據庫，視作穩定節點：
for _, ch := range waiting {
close(ch)
}
waiting, refreshDone = nil, nil
case <-refresh.C: //定時刷新k桶事件,refreshInterval=30 min
revalidateDone = make(chan struct{})
go tab.doRevalidate(revalidateDone)
case req := <-tab.refreshReq: //刷新k桶的請求事件
revalidate.Reset(tab.nextRevalidateTime())
revalidateDone = nil
case <-refreshDone: // doRefresh完成
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 <-revalidate.C: // 驗證k桶節點有效性，10 second
case <-revalidateDone: // 驗證K桶節點有效性完成
// 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
}
case <-copyNodes.C: // 定時(30秒)將節點存入數據庫，如果某個節點在k桶中存在超過5分鐘，則認為它是一個穩定的節點
}節點查找
// 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()
}
}
getNode函數用於根據ID來查找節點，如果不存在則返回nil：
// 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
}
節點發現
// 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
}
以太坊分佈式網絡採用了結構化網絡模型，其實現方案使用Kademlia協議，下面我們對節點發現進行簡單介紹，在以太坊中k值是16，也就是說每個k桶包含16個節點，一共256個k桶，K桶中記錄節點的NodeId，Distance，Endpoint，IP等信息，並按照與Target節點的距離排序，節點查找由doRefresh()實現：
// 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
}
從上述代碼中可以看到這里首先調用tab.loadSeedNodes()從數據庫中加載節點並將其插入到表中去：
之後通過lookupSelf來發現新的節點，這裡會優先使用當前節點的ID來運行newLookup發現鄰居節點：
// 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
}
最後隨機一個target，進行lookup:
// 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
}
#####
服務結構
server端的的數據結構如下所示：
// 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():
}
}
Server配置(本地節點秘鑰、撥號比率、節點最大鍊接數、撥號比率、事件記錄等)：
// 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:
}
}
新增節點
AddPeer函數用於新增一個給定的節點，其實現代碼如下所示：
// 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():
}
}
AddTrustedPeer函數用於新增一個可信任節點：
// 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:
}
}
移除節點
RemovePeer函數用於移除節點並斷開與節點之間的連接：
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()
}
RemoveTrustedPeer函數用於移除一個可信任節點：
終止服務
// 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
}
Stop函數用於終止節點運行，具體代碼如下所示：
// 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{})
服務啟動
位於go-ethereum-1.10.2\p2p\server.go中的start函數用於啟動一個P2P節點：
srv.setupDialScheduler()
srv.loopWG.Add(1)
go srv.run()
return nil
}
在這里首先檢查當前節點是否處於運行狀態，如果是則直接返回並給出錯誤提示信息，如果不是則將srv.running設置為true，之後進入服務啟動流程，之後檢查log是否開啟等，之後初始化配置P2P服務信息：
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)
}
}
之後調用setupLocalNode來啟動一個本地節點，並建立本地監聽，然後配置一個DiscoveryV5網絡協議，生成節點路由表。
// 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
}
之後調用setupDialScheduler啟動主動撥號連接過程，然後開一個協程，在其中做peer的維護：
setupDialScheduler代碼如下所示，這里通過newDialScheduler來建立連接，參數discmix確定了進行主動建立連接時的節點集，它是一個迭代器，同時將setupConn連接建立函數傳入：
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
}
newDialScheduler函數如下所示，在這里通過d.readNodes(it)從迭代器中取得節點，之後通過通道傳入d.loop(it)中進行連接：
// 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{}{}
}()
}
}
服務監聽
// 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
}
在上面的服務啟動過程中有一個setupListening函數，該函數用於監聽事件，具體代碼如下所示：
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.listenLoop()，該函數是一個死循環的goroutine，它會監聽端口並接收外部的請求：
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,這裡的SetupConn主要執行執行握手協議，並嘗試把鏈接創建為一個peer對象：
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,在上述代碼中又去調用了srv.setupConn(c, flags, dialDest)函數，該函數用於執行握手協議:
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,if dialDest != nil { // dest=nil 被動連接，dest!=nil主動連接誒
if err != nil {
clog.Trace("Rejected peer", "err", err)
return err
}
return nil
}
err := c.doEncHandshake(srv.PrivateKey) // 公鑰交換，確定共享秘鑰RLPx層面的握手一來一去
// 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)
}
err := c.doProtoHandshake(srv.ourHandshake) // 進行協議層面的握手,也即p2p握手，一來一去
// 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
)
srv.checkpointAddPeer) // 狀態校驗
sec, err = initiatorEncHandshake(c.conn, prv,秘鑰握手通過deEncHandshake函數實現，在函數之中調用了Handshake()函數：
Handshake代碼如下所示，在這裡會根據是主動握手還是被動握手來進行執行對應的握手邏輯：
sec, err = receiverEncHandshake(c.conn, prv)
}
if err != nil {
return nil, err
}
c.InitWithSecrets(sec)
return sec.remote, err
}
if c.dialDest != nil { //主動握手
// 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)
}
c.dialDest) //主動發起秘鑰驗證握手結束，確定共享秘鑰
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)
}
} else { // 被動握手
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
}
主動發起握手過程過程如下，在這裡會調用makeAuthMsg來生成Auth身份信息，包含簽名，隨機nonce生成的與簽名對應的公鑰和版本號，之後調用sealEIP8方法進行rlpx編碼，之後發起加密握手，之後接收返回的authResp消息，並驗證解密，獲取對方公鑰，之後生成AES，MAC：
receiverEncHandshake如下所示，和initiatorEncHandshake相差無幾：
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())
}
}
之後通過doProtoHandshake來完成協議握手操作，在這裡調用send發送一次握手操作，之後通過readProtocolHandshake來讀取返回信息，之後進行檢查：
服務循環
// 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
