深入区块链以太坊源码之p2p通信
Mohammad61417
发表于 2022-12-7 15:31:22
133
0
0
无结构化的:. B- d2 B$ V' H/ p9 p6 T6 \2 B
这种p2p网络即最普通的,不对结构作特别设计的实现方案。
优点是结构简单易于组建,网络局部区域内个体可任意分布,9 N5 l* {5 a9 @* L. ?; @7 \- Q
反正此时网络结构对此也没有限制;特别是在应对大量新个体加- S3 N3 e& X* ~ ^
入网络和旧个体离开网络(“churn”)时它的表现非常稳定。
缺点在于在该网络中查找数据的效率太低,因为没有预知信息,* E6 ~; R8 m4 e5 f3 t
所以往往需要将查询请求发遍整个网络(至少大多数个体),
这会占用很大一部分网络资源,并大大拖慢网络中其他业务运行。/ E5 x3 m9 p. j d2 \" f
结构化的:
这种p2p网络中的个体分布经过精心设计,主要目的是为了提高查询数据的效率,- X4 s; r9 L' ]+ G1 e
降低查询数据带来的资源消耗。
以太坊采用了不需要结构化的结构,经过改进的非结构化(比如设计好相邻个体列表peerSet结构)
网络模型可以满足需求;
二、分布式hash表(DHT)0 C: i# {. S* K* g- }( o2 w
保存数据! v5 M1 I9 q, g0 g
(以下只是大致原理,具体的协议实现可能会有差异)
当某个节点得到了新加入的数据(K/V),它会先计算自己与新数据的 key 之间的“距离”;
然后再计算它所知道的其它节点与这个 key 的距离。
如果计算下来,自己与 key 的距离最小,那么这个数据就保持在自己这里。! A) `. q; j8 w' t% ?8 {+ M6 [! R& F
否则的话,把这个数据转发给距离最小的节点。
收到数据的另一个节点,也采用上述过程进行处理(递归处理)。
获取数据% d; S0 D! ~/ p* S& b- V: v( Z
(以下只是大致原理,具体的协议实现可能会有差异)! J3 X4 @/ S8 h* Z* H' Y- i
当某个节点接收到查询数据的请求(key),它会先计算自己与 key 之间的“距离”;+ V/ B3 |4 o7 J) f) d2 O* y) Z
然后再计算它所知道的其它节点与这个 key 的距离。/ a" F5 Q, A& f4 K! Z
如果计算下来,自己与 key 的距离最小,那么就在自己这里找有没有 key 对应的 value。
有的话就返回 value,没有的话就报错。
否则的话,把这个数据转发给距离最小的节点。
收到数据的另一个节点,也采用上述过程进行处理(递归处理)。; E+ q; k( u5 S
三、以太坊中p2p通信的管理模块ProtocolManager& G4 c* I3 i1 k$ u3 O+ u$ V
/geth.go* n2 ^% }# s1 Y9 U5 }
// Start creates a live P2P node and starts running it.; @2 |2 [' ?% c( ]# P y
func (n *Node) Start() error {- s6 d7 ^. G% S
return n.node.Start()" v& J! V, G6 o, c+ c4 R" U9 [
}& T! ]& T3 J9 T+ T0 ]0 }
/*5 F/ \& ^/ J) v4 A2 z3 ^8 c7 C
Protocol:容纳应用程序所要求的回调函数等.并通过p2p.Server{}在新连接建立后,将其传递给通信对象peer。) o5 U2 T1 b4 h3 Q5 Y
Node.Start()中首先会创建p2p.Server{},此时Server中的Protocol[]还是空的;( h# Y, U$ k! E8 X( _
然后将Node中载入的所有实现体中的Protocol都收集起来,
一并交给Server对象,作为Server.Protocols列表;然后启动Server对象,
并将Server对象作为参数去逐一启动每个实现体。
*/) T9 C* B# q u p) |2 k/ {
/node.go
// Start create a live P2P node and starts running it.6 I, R: U6 J) W7 Q2 X
func (n *Node) Start() error {
...3 |4 E4 j2 i- C I* s: i& X, i
/*
...( x" I/ @1 K) x N, T( e- [) y
初始化serverConfig
*/
running := &p2p.Server{Config: n.serverConfig}6 t9 N0 f, s8 m: { E2 B
...
// Gather the protocols and start the freshly assembled P2P server
for _, service := range services {, m1 L0 _. t5 j3 D) y1 W2 [3 P0 S
running.Protocols = append(running.Protocols, service.Protocols()...)
}: W% p2 _# Y2 c' r- _3 a8 h% c0 g% g
if err := running.Start(); err != nil { //见下面的(srv *Server)Start方法, p9 Y# d+ n: v9 q' |
return convertFileLockError(err)
}; q( H Q! F% b7 E- M
// Start each of the services
started := []reflect.Type{}" ]% \2 M% h* {
for kind, service := range services {; U4 y- }% [, K8 Y6 m+ H: {% e1 m7 A$ h
// Start the next service, stopping all previous upon failure
//启动每个services通过下面的方法func (s *Ethereum) Start(srvr *p2p.Server) error {
if err := service.Start(running); err != nil {
for _, kind := range started {0 d6 V# ^3 W8 `' \0 z" B# X
services[kind].Stop()
}* ]' _7 w5 C# F. l0 G
running.Stop()3 s+ |5 {+ P+ {' V
return err" T$ I" O% b% Y4 J( B, {1 }
}4 u0 S) `$ D3 P! E3 N1 `
...
}
}$ h, F2 n5 c% C* \7 ?
// Start starts running the server.
// Servers can not be re-used after stopping.0 n8 S2 R5 S( Q# ?% w
func (srv *Server) Start() (err error) {
srv.lock.Lock()
//srv.lock为了避免多线程重复启动
defer srv.lock.Unlock()
if srv.running {
return errors.New("server already running"): s( ]7 J$ x% l" U
}
srv.running = true$ O' k) L+ ~+ l. n7 M2 x
srv.log = srv.Config.Logger- k0 ?" i% C U+ g% j3 W* N/ N
if srv.log == nil {" z/ ?' \1 u$ `
srv.log = log.New()
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}+ a: k. O' |* w' l' U
// static fields
if srv.PrivateKey == nil {* w: N) Z8 C1 `
return fmt.Errorf("Server.PrivateKey must be set to a non-nil key")6 A6 O. \0 r4 a& l8 s; f! v0 L
} U2 l/ [- U' p) W- l' q a1 A6 d
//newTransport使用了newRLPX使用了rlpx.go中的网络协议。
if srv.newTransport == nil {
srv.newTransport = newRLPX
}( B5 f6 H1 c# N( h
if srv.Dialer == nil {
srv.Dialer = TCPDialer{&net.Dialer{Timeout: defaultDialTimeout}}. A8 K8 x+ K' g6 o3 [
}
srv.quit = make(chan struct{}) P; d5 ?: Q7 c5 l! E
srv.addpeer = make(chan *conn)7 y, b6 }' l7 c% C
srv.delpeer = make(chan peerDrop)
srv.posthandshake = make(chan *conn)
srv.addstatic = make(chan *enode.Node)
srv.removestatic = make(chan *enode.Node)
srv.addtrusted = make(chan *enode.Node)0 P* U+ g% U0 @% o+ V4 G" {
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
//srv.setupLocalNode()这里主要执行握手, k6 l/ u) V/ e# J3 i$ v
if err := srv.setupLocalNode(); err != nil {) O6 f5 {7 e/ g2 B8 U+ g. L% w+ S
return err* R( \7 p- h! \* a4 d% |
}7 \3 Z. l: w8 o9 F0 c
if srv.ListenAddr != "" {
//监听TCP端口-->用于业务数据传输,基于RLPx协议)/ P) y1 I4 j7 l/ z2 V- L+ P
//在setupListening中有个go srv.listenLoop()去监听某个端口有无主动发来的IP连接/ t1 |: h1 h; j- a& Q3 q) o7 ?" w; d
if err := srv.setupListening(); err != nil {
return err
}
}
//侦听UDP端口(用于结点发现内部会启动goroutine)
if err := srv.setupDiscovery(); err != nil {9 r0 R s! ]9 \( }. ^" X
return err
}
dynPeers := srv.maxDialedConns()+ g1 ]6 I3 Z# ~# ^# y
dialer := newDialState(srv.localnode.ID(), srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict)" g" O. H0 n/ i
srv.loopWG.Add(1)
// 启动新线程发起TCP连接请求
//在run()函数中,监听srv.addpeer通道有没有信息如果有远端peer发来连接请求,0 V8 S4 E- ?6 A
//则调用Server.newPeer()生成新的peer对象,并把Server.Protocols全交给peer。* m; h2 F0 H& o Q% `+ l- T1 N
/*; u g8 R/ ~, i$ M) Z
case c := 0 {
if s.config.LightPeers >= srvr.MaxPeers {
return fmt.Errorf("invalid peer config: light peer count (%d) >= total peer count (%d)", s.config.LightPeers, srvr.MaxPeers)
}
maxPeers -= s.config.LightPeers. r1 Z) y# j8 I- Y5 J
}+ {, M2 g) W- ~, W
// Start the networking layer and the light server if requested* L2 K- S1 q- D7 b% f1 G; r
s.protocolManager.Start(maxPeers)
if s.lesServer != nil {1 T' \( `1 G+ J
s.lesServer.Start(srvr)$ \9 i! e0 G+ h
}! i- M0 b- [4 W
return nil" Z" A- A+ |9 U
}* j; Y" k# E7 l1 ~+ m: l1 m
/eth/handler.go# H$ w5 y7 V9 p! P5 Z
type ProtocolManager struct {
networkID uint64, `* H" C1 p# N
fastSync uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks)0 U, f" {/ P3 T$ F" I* Z" B/ o
acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing)
txpool txPool0 z U( D0 o! n7 d; Q- w8 c
blockchain *core.BlockChain6 e8 ~! G3 R3 J( a- g A2 Y& O
chainconfig *params.ChainConfig
maxPeers int
//Downloader类型成员负责所有向相邻个体主动发起的同步流程。+ }* r) e/ }% Q2 H! Q1 T
downloader *downloader.Downloader
//Fetcher类型成员累积所有其他个体发送来的有关新数据的宣布消息,并在自身对照后做出安排
fetcher *fetcher.Fetcher; u" W" s$ Q5 ~/ C
//用来缓存相邻个体列表,peer{}表示网络中的一个远端个体。
peers *peerSet$ g7 v9 z( \ ]1 q
SubProtocols []p2p.Protocol
eventMux *event.TypeMux
txsCh chan core.NewTxsEvent
txsSub event.Subscription
minedBlockSub *event.TypeMuxSubscription) e+ v+ ~) e! n( }8 V
//通过各种通道(chan)和事件订阅(subscription)的方式,接收和发送包括交易和区块在内的数据更新。
//当然在应用中,订阅也往往利用通道来实现事件通知。
// channels for fetcher, syncer, txsyncLoop
newPeerCh chan *peer6 O: d9 F- t. c
txsyncCh chan *txsync" ^; c1 k' \/ z5 ~, d2 n' O$ _
quitSync chan struct{}
noMorePeers chan struct{}* ], u% y, S. G9 A/ G
// wait group is used for graceful shutdowns during downloading! c2 |- e6 n& X8 O
// and processing
wg sync.WaitGroup
}
Start()函数是ProtocolManager的启动函数,它会在eth.Ethereum.Start()中被主动调用。, o8 F2 U, R& w$ {; K1 T
ProtocolManager.Start()会启用4个单独线程(goroutine,协程)去分别执行4个函数," T5 U5 f0 {0 _6 s7 k' O, H4 p
这也标志着该以太坊个体p2p通信的全面启动。
func (pm *ProtocolManager) Start(maxPeers int) {2 l- n- Q+ @, | s- W. t% H* v( l1 C
pm.maxPeers = maxPeers
// broadcast transactions
//广播交易的通道。 txsCh会作为txpool的TxPreEvent订阅通道。: r$ Q( n6 Q3 m$ {- H; B" J
//txpool有了这种消息会通知给这个txsCh。 广播交易的goroutine会把这个消息广播出去。
pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
//订阅交易信息/ j5 x' ^2 A% M$ \( r1 t- W
pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)
go pm.txBroadcastLoop()6 h6 n3 w* X4 }$ N1 Y2 d
//订阅挖矿消息。当新的Block被挖出来的时候会产生消息; ]5 y8 W* [: ^
// broadcast mined blocks: R4 {3 H8 u9 P
pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
//挖矿广播 goroutine 当挖出来的时候需要尽快的广播到网络上面去。
go pm.minedBroadcastLoop()
// start sync handlers
// 同步器负责周期性地与网络同步,下载散列和块以及处理通知处理程序。/ A% G9 ~4 f& G
go pm.syncer()
// txsyncLoop负责每个新连接的初始事务同步。 当新的peer出现时,
// 转发所有当前待处理的事务。为了最小化出口带宽使用,我们一次只发送一个小包。
go pm.txsyncLoop()
}) j( o* {& ~, j& c* `: O% K2 @
//txBroadcastLoop()会在txCh通道的收端持续等待,一旦接收到有关新交易的事件,7 p* P( e/ ~4 `7 ^
//会立即调用BroadcastTx()函数广播给那些尚无该交易对象的相邻个体。
//------------------go pm.txBroadcastLoop()-----------------------+ r0 `3 {$ n9 z1 q. U/ I" j4 {& Z) [
func (pm *ProtocolManager) txBroadcastLoop() {
for { z2 M/ ~& R# ?* ?" O
select {
case event := = pm.maxPeers && !p.Peer.Info().Network.Trusted {
return p2p.DiscTooManyPeers
}
p.Log().Debug("Ethereum peer connected", "name", p.Name())$ x/ i1 }2 f, `, @
// Execute the Ethereum handshake- x. S& O5 o( Z9 U% p/ f& p% X
var (
genesis = pm.blockchain.Genesis()
head = pm.blockchain.CurrentHeader(); W4 l. J: [! n: U* H+ ?' j! e
hash = head.Hash()" h% }% U$ J: g+ }
number = head.Number.Uint64()
td = pm.blockchain.GetTd(hash, number)' X" _9 {3 y. J: }5 O6 e
)
//握手,与对方peer沟通己方的区块链状态
if err := p.Handshake(pm.networkID, td, hash, genesis.Hash()); err != nil {
p.Log().Debug("Ethereum handshake failed", "err", err)
return err5 P0 E4 p1 [4 o& C, k( e* F
}: a* [5 m) H1 L) @6 j
//初始化一个读写通道,用以跟对方peer相互数据传输。
if rw, ok := p.rw.(*meteredMsgReadWriter); ok {: m @# ]8 P/ C. e ^9 w% x
rw.Init(p.version)# E1 y& s2 {3 l5 C* R
}* r9 B) C1 e3 |2 b* \. w
// Register the peer locally
//注册对方peer,存入己方peer列表;只有handle()函数退出时,才会将这个peer移除出列表。" E# N$ ~% p( w9 P, R
if err := pm.peers.Register(p); err != nil {; e' R' T# [- A8 b: e# x- |% \ B
p.Log().Error("Ethereum peer registration failed", "err", err)) u O$ ^; [; Y
return err
}
defer pm.removePeer(p.id)
//Downloader成员注册这个新peer;Downloader会自己维护一个相邻peer列表。2 K$ e: h7 ]7 S i% I
// Register the peer in the downloader. If the downloader considers it banned, we disconnect
if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
return err
}7 I" V% B0 d# T+ ?2 u
// Propagate existing transactions. new transactions appearing
// after this will be sent via broadcasts.* f" p3 S2 H1 s9 V& l& u0 H0 D- V
/*
调用syncTransactions(),用当前txpool中新累计的tx对象组装成一个txsync{}对象,- F9 l% ?8 G7 c/ s; J
推送到内部通道txsyncCh。还记得Start()启动的四个函数么? 其中第四项txsyncLoop()# l3 b; ~) D+ U5 V
中用以等待txsync{}数据的通道txsyncCh,正是在这里被推入txsync{}的。7 B9 D- G( l# C; \; W, I% }
*/. [$ ~' C. p9 [/ k' P: Y) v' r% _
pm.syncTransactions(p)
// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork# n* P; r8 w9 k) `
if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {/ G4 H. k" J# g. Q. B
// Request the peer's DAO fork header for extra-data validation
if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
return err5 j1 M! K9 o3 }
}! o, c, G0 r- v! [/ y
// Start a timer to disconnect if the peer doesn't reply in time
p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {5 t# H3 S5 m) q! f' y3 _/ U# H. H
p.Log().Debug("Timed out DAO fork-check, dropping")
pm.removePeer(p.id)
})
// Make sure it's cleaned up if the peer dies off
defer func() {/ U! i. Q: Y( ~4 [( _) k5 B/ E
if p.forkDrop != nil {' p6 b5 K. ^+ J" c
p.forkDrop.Stop()1 h, |, y+ B3 { F6 `$ `" g
p.forkDrop = nil
}
}()
}. S. E" W! @$ f4 O- w' D
//在无限循环中启动handleMsg(),当对方peer发出任何msg时," C1 @7 ^7 n8 I/ _ ^! _* q; O( ~
//handleMsg()可以捕捉相应类型的消息并在己方进行处理。2 |# u) I9 r" F
// main loop. handle incoming messages.
for {
if err := pm.handleMsg(p); err != nil {
p.Log().Debug("Ethereum message handling failed", "err", err)
return err
}
}/ e2 F8 h9 N$ G0 ~9 C$ u5 F
}
成为第一个吐槽的人