深入区块链以太坊源码之p2p通信
Mohammad61417
发表于 2022-12-7 15:31:22
127
0
0
无结构化的:
这种p2p网络即最普通的,不对结构作特别设计的实现方案。% S I/ o# u# O/ X* H1 @$ {
优点是结构简单易于组建,网络局部区域内个体可任意分布,
反正此时网络结构对此也没有限制;特别是在应对大量新个体加
入网络和旧个体离开网络(“churn”)时它的表现非常稳定。
缺点在于在该网络中查找数据的效率太低,因为没有预知信息,
所以往往需要将查询请求发遍整个网络(至少大多数个体),
这会占用很大一部分网络资源,并大大拖慢网络中其他业务运行。& C( t0 u4 E$ H% d* Z
结构化的:# L1 k: ]8 D9 C/ B! M6 w
这种p2p网络中的个体分布经过精心设计,主要目的是为了提高查询数据的效率,
降低查询数据带来的资源消耗。
以太坊采用了不需要结构化的结构,经过改进的非结构化(比如设计好相邻个体列表peerSet结构)4 S$ E& J1 e; Q$ c- e' `$ U
网络模型可以满足需求;
二、分布式hash表(DHT)# L/ c4 r/ p) T
保存数据
(以下只是大致原理,具体的协议实现可能会有差异)0 I! B2 t, ^+ B1 X$ n8 Z0 g- m
当某个节点得到了新加入的数据(K/V),它会先计算自己与新数据的 key 之间的“距离”;
然后再计算它所知道的其它节点与这个 key 的距离。
如果计算下来,自己与 key 的距离最小,那么这个数据就保持在自己这里。
否则的话,把这个数据转发给距离最小的节点。
收到数据的另一个节点,也采用上述过程进行处理(递归处理)。5 k* r2 h e/ }. Q9 ~& R6 Q' c6 F
获取数据
(以下只是大致原理,具体的协议实现可能会有差异)
当某个节点接收到查询数据的请求(key),它会先计算自己与 key 之间的“距离”; k9 P) F1 p! r0 h/ ~* m! j9 j
然后再计算它所知道的其它节点与这个 key 的距离。
如果计算下来,自己与 key 的距离最小,那么就在自己这里找有没有 key 对应的 value。& F+ ], n5 r& {: ~
有的话就返回 value,没有的话就报错。
否则的话,把这个数据转发给距离最小的节点。
收到数据的另一个节点,也采用上述过程进行处理(递归处理)。
三、以太坊中p2p通信的管理模块ProtocolManager" Z8 T4 B/ K) d3 p( I; o
/geth.go
// Start creates a live P2P node and starts running it.! s+ [$ N7 P( i, F
func (n *Node) Start() error {/ _- G" |9 d* r
return n.node.Start()
}
/*
Protocol:容纳应用程序所要求的回调函数等.并通过p2p.Server{}在新连接建立后,将其传递给通信对象peer。& T5 U% B! _2 Y# ^$ E( }5 L4 @
Node.Start()中首先会创建p2p.Server{},此时Server中的Protocol[]还是空的;; R2 e Z, Q) g
然后将Node中载入的所有实现体中的Protocol都收集起来,7 p: t B( u o: S j
一并交给Server对象,作为Server.Protocols列表;然后启动Server对象,
并将Server对象作为参数去逐一启动每个实现体。
*/
/node.go2 @3 k, j0 w' n' }! ]" j; C
// Start create a live P2P node and starts running it.0 K' g+ a0 V3 s
func (n *Node) Start() error {8 h! f. l3 g0 [
6 R' p& \2 m( s! J; q% y4 r
...
/*
...
初始化serverConfig6 E* m! G, i' D% K
*/! t8 w2 N6 l K+ M( W6 P- v+ Q3 V
running := &p2p.Server{Config: n.serverConfig}3 `1 _6 n2 y7 D* Z, C4 d' q+ g
...0 i( I9 B. U" q# |" ~
// Gather the protocols and start the freshly assembled P2P server
for _, service := range services {5 m. ]# J/ B1 ?& E% H$ d% P' V; ?
running.Protocols = append(running.Protocols, service.Protocols()...)0 P& _( }3 }9 J/ V; u
}+ E4 n6 N. P( H( U1 `. P C
if err := running.Start(); err != nil { //见下面的(srv *Server)Start方法
return convertFileLockError(err)- b7 o! d# R9 t4 B
}
// Start each of the services# H) P9 y2 j, k q6 ]$ V" z
started := []reflect.Type{}
for kind, service := range services {; t* p5 `/ {! y
// 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 {2 v+ O4 F' c& k8 n8 ]7 U& @
services[kind].Stop()
}6 z. o, O1 j* q0 ~1 T# r4 s
running.Stop()
return err7 ?8 J& P0 X! J( P% z4 w* e0 w- D3 ~
}
...
} K% G1 R9 ?8 K- [+ d% |' _+ k
}" a5 t) s# L7 k: m$ x9 r
// Start starts running the server.
// Servers can not be re-used after stopping.
func (srv *Server) Start() (err error) {
srv.lock.Lock()- }0 z9 u8 e/ B& J5 I
//srv.lock为了避免多线程重复启动/ A% ~, _% ]# Y
defer srv.lock.Unlock()
if srv.running {6 v# `. w Y$ m v
return errors.New("server already running")3 b& O/ F0 }9 _' X. E4 K
}
srv.running = true
srv.log = srv.Config.Logger" v$ j" R' c; R9 y. N7 P3 s) W6 B
if srv.log == nil {
srv.log = log.New()
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}, n- y* d( C1 k1 l! [
// static fields! j) `/ q- C) q: N6 m
if srv.PrivateKey == nil {4 Y7 v: B& f u- W6 S; k0 }
return fmt.Errorf("Server.PrivateKey must be set to a non-nil key")) {% Q: s7 s( z0 I! r# S
}! R& ]/ y! T% C& s! X
//newTransport使用了newRLPX使用了rlpx.go中的网络协议。
if srv.newTransport == nil {
srv.newTransport = newRLPX
}7 A$ u/ W3 H) C9 r3 Y2 \- z* X8 ~- W
if srv.Dialer == nil {
srv.Dialer = TCPDialer{&net.Dialer{Timeout: defaultDialTimeout}}
}) c8 g7 I, E6 `5 l- p; h
srv.quit = make(chan struct{})4 K3 S2 _6 V6 Y- A0 _+ j% x
srv.addpeer = make(chan *conn)" v$ n _4 w% V/ e) b1 y/ q3 J! D
srv.delpeer = make(chan peerDrop)
srv.posthandshake = make(chan *conn)+ T; J& y% b. { ^' s
srv.addstatic = make(chan *enode.Node)
srv.removestatic = make(chan *enode.Node); f1 F/ w _5 Z) b
srv.addtrusted = make(chan *enode.Node)
srv.removetrusted = make(chan *enode.Node)
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
//srv.setupLocalNode()这里主要执行握手
if err := srv.setupLocalNode(); err != nil {, y; O6 `$ W$ ^; ]- @" A
return err
}% i2 I7 J$ w% o: O# i8 [
if srv.ListenAddr != "" { _, e& K; y5 `) i/ @! f# W0 L( z
//监听TCP端口-->用于业务数据传输,基于RLPx协议)
//在setupListening中有个go srv.listenLoop()去监听某个端口有无主动发来的IP连接
if err := srv.setupListening(); err != nil {
return err
}
}3 i0 e- a t5 N; b- \# U" M
//侦听UDP端口(用于结点发现内部会启动goroutine)% @& D2 A" S/ y& Y% f
if err := srv.setupDiscovery(); err != nil {
return err! Z' q/ \' ~- u
}- m. w9 H/ z8 z& E& Q% H5 h" x
dynPeers := srv.maxDialedConns()" ~8 O7 J D4 v9 L! F, x- s: k
dialer := newDialState(srv.localnode.ID(), srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict)
srv.loopWG.Add(1)
// 启动新线程发起TCP连接请求1 z1 e* q. d" [
//在run()函数中,监听srv.addpeer通道有没有信息如果有远端peer发来连接请求," G; R% D& F; T, j p+ M. ?% R
//则调用Server.newPeer()生成新的peer对象,并把Server.Protocols全交给peer。3 O; P" ]# Q8 N+ e
/*" d) F& E1 P6 T2 k j
case c := 0 {" d y1 J; V4 a
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)
}% S2 F; J8 ]9 ~( `/ ~1 B* n
maxPeers -= s.config.LightPeers
} v6 e/ A# n/ I- c% |1 ^
// Start the networking layer and the light server if requested3 o; e3 e; Y! k8 Y
s.protocolManager.Start(maxPeers)% H! S" K v# @; X0 W2 Y* T
if s.lesServer != nil {
s.lesServer.Start(srvr)
}
return nil7 _: `( U- ~+ Z- H3 U
}
/eth/handler.go3 |1 D! f/ e' m' Y L4 r1 g
type ProtocolManager struct {
networkID uint64+ s2 W$ H; e& {3 l
fastSync uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks)
acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing)$ w$ @" d& ?+ `/ y
txpool txPool
blockchain *core.BlockChain
chainconfig *params.ChainConfig
maxPeers int* t- w+ k$ _* T. W4 K. ~, k
//Downloader类型成员负责所有向相邻个体主动发起的同步流程。; }, }1 @3 a6 @4 x2 g
downloader *downloader.Downloader5 x- f' E) f. s; r
//Fetcher类型成员累积所有其他个体发送来的有关新数据的宣布消息,并在自身对照后做出安排( ^" S2 Q! a0 \5 u
fetcher *fetcher.Fetcher+ F6 {( I K) n: I; S( b9 i n) c" j
//用来缓存相邻个体列表,peer{}表示网络中的一个远端个体。
peers *peerSet
SubProtocols []p2p.Protocol2 `1 U Z; \! ]4 Q
eventMux *event.TypeMux# A V! Z$ U9 E- K7 @) H a4 V
txsCh chan core.NewTxsEvent
txsSub event.Subscription6 D/ y* y; f5 M h! q" G& R
minedBlockSub *event.TypeMuxSubscription9 M; Y2 U8 j( K+ h+ V N. P5 d
* `% e2 b! N2 C+ O
//通过各种通道(chan)和事件订阅(subscription)的方式,接收和发送包括交易和区块在内的数据更新。* |4 d, x( w% [& s
//当然在应用中,订阅也往往利用通道来实现事件通知。4 K/ f" k: @: @8 o
// channels for fetcher, syncer, txsyncLoop
newPeerCh chan *peer' Y& n; i& ~5 h4 D
txsyncCh chan *txsync
quitSync chan struct{}
noMorePeers chan struct{}4 v D- k/ f, T( I3 m9 Y
// wait group is used for graceful shutdowns during downloading- Q# S# P7 m" s! k; l! d! S0 t& Q3 d
// and processing( j. |' i0 l! ^& |+ K
wg sync.WaitGroup. D5 n+ f2 i6 P; f
}
Start()函数是ProtocolManager的启动函数,它会在eth.Ethereum.Start()中被主动调用。5 u, c2 H8 T* ~+ T8 |0 @7 P9 g, |
ProtocolManager.Start()会启用4个单独线程(goroutine,协程)去分别执行4个函数,+ e2 |$ `9 p0 ?5 H* \
这也标志着该以太坊个体p2p通信的全面启动。' w- l, ~& b/ m3 s: v: i3 Y
func (pm *ProtocolManager) Start(maxPeers int) {
pm.maxPeers = maxPeers4 a) p, q" Q- D& ]
// broadcast transactions
//广播交易的通道。 txsCh会作为txpool的TxPreEvent订阅通道。
//txpool有了这种消息会通知给这个txsCh。 广播交易的goroutine会把这个消息广播出去。& l+ z% R# D( Y' X8 b
pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
//订阅交易信息. Y5 C* v, a5 |6 c+ P/ b& |
pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)6 i4 j1 ^9 e6 C: p' b3 v3 p
" j7 K" ?& h4 A4 w0 f
go pm.txBroadcastLoop()
//订阅挖矿消息。当新的Block被挖出来的时候会产生消息, Q; Y+ s" m# ~9 W" w
// broadcast mined blocks
pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
//挖矿广播 goroutine 当挖出来的时候需要尽快的广播到网络上面去。
go pm.minedBroadcastLoop()3 ^" N) O8 J3 w+ N7 v" R# J
// start sync handlers
// 同步器负责周期性地与网络同步,下载散列和块以及处理通知处理程序。. {# C9 k1 E! J
go pm.syncer()/ P% Q: P( {- T
// txsyncLoop负责每个新连接的初始事务同步。 当新的peer出现时,' {) T$ H# z* J0 D+ S$ K
// 转发所有当前待处理的事务。为了最小化出口带宽使用,我们一次只发送一个小包。
go pm.txsyncLoop()& _# J2 D/ k+ ?/ }/ H( ^: p
}
//txBroadcastLoop()会在txCh通道的收端持续等待,一旦接收到有关新交易的事件,
//会立即调用BroadcastTx()函数广播给那些尚无该交易对象的相邻个体。. b4 J) u$ V( X6 w! N" ~
//------------------go pm.txBroadcastLoop()-----------------------# y" D- O0 F, `
func (pm *ProtocolManager) txBroadcastLoop() {; r$ O( b$ o \% t; a6 \
for {
select {6 K \; a$ M, A6 G% I+ K
case event := = pm.maxPeers && !p.Peer.Info().Network.Trusted {
return p2p.DiscTooManyPeers
}
p.Log().Debug("Ethereum peer connected", "name", p.Name())
// Execute the Ethereum handshake2 B! j/ N# Y9 l
var (
genesis = pm.blockchain.Genesis()9 L! M, F1 E" H$ w2 L
head = pm.blockchain.CurrentHeader()
hash = head.Hash()
number = head.Number.Uint64()
td = pm.blockchain.GetTd(hash, number)3 _& ~( ^; J/ s/ S; k: a* y
)
//握手,与对方peer沟通己方的区块链状态% ]$ j4 c" r7 u& m' h6 b
if err := p.Handshake(pm.networkID, td, hash, genesis.Hash()); err != nil {
p.Log().Debug("Ethereum handshake failed", "err", err)
return err0 J) D2 j0 W4 s( z6 x+ p
}; ~) V. B* s& m
//初始化一个读写通道,用以跟对方peer相互数据传输。
if rw, ok := p.rw.(*meteredMsgReadWriter); ok {
rw.Init(p.version)
}
// Register the peer locally
//注册对方peer,存入己方peer列表;只有handle()函数退出时,才会将这个peer移除出列表。
if err := pm.peers.Register(p); err != nil {
p.Log().Error("Ethereum peer registration failed", "err", err) u9 l; o9 B. @. b
return err$ x9 J( f% G- ^4 u+ L
}, F! a/ i# y+ }1 A6 L
defer pm.removePeer(p.id)6 o- }. E4 V0 v$ G5 M! f
//Downloader成员注册这个新peer;Downloader会自己维护一个相邻peer列表。$ N V, y- C) j
// 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, u5 C) i1 }. ~$ r9 O
}8 o9 [" [' ]2 l' i- g: N4 X: d, b
// Propagate existing transactions. new transactions appearing
// after this will be sent via broadcasts.
/*
调用syncTransactions(),用当前txpool中新累计的tx对象组装成一个txsync{}对象,; L; e8 Y2 @- o* }
推送到内部通道txsyncCh。还记得Start()启动的四个函数么? 其中第四项txsyncLoop()
中用以等待txsync{}数据的通道txsyncCh,正是在这里被推入txsync{}的。
*/
pm.syncTransactions(p)0 h7 O X0 L6 h6 l
// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork3 ^% g! D+ j r3 ^% ]
if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {; x0 q f( h( Y
// Request the peer's DAO fork header for extra-data validation
if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
return err+ O3 ]0 x ]' Q5 l) |
}
// Start a timer to disconnect if the peer doesn't reply in time
p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {
p.Log().Debug("Timed out DAO fork-check, dropping")" Z/ e% j3 C p ^
pm.removePeer(p.id)* f# S! Z. Q$ a4 X3 | ?! _% N
})2 P! |' b) j" x6 N
// Make sure it's cleaned up if the peer dies off
defer func() {8 r5 N1 `& s6 E: h& A* k3 U' a% G
if p.forkDrop != nil {
p.forkDrop.Stop()& ]# G% C/ G% {4 |& L+ `$ R
p.forkDrop = nil
}
}(). b" }% d- ~9 \6 w+ ~& [1 V: C Y
} [8 ^7 K+ ~4 B$ X1 A! s
//在无限循环中启动handleMsg(),当对方peer发出任何msg时,
//handleMsg()可以捕捉相应类型的消息并在己方进行处理。9 O& [1 N+ Z5 ]1 i- k) j; l9 u4 ]5 \
// main loop. handle incoming messages.
for {
if err := pm.handleMsg(p); err != nil {
p.Log().Debug("Ethereum message handling failed", "err", err)
return err
}
}
}
成为第一个吐槽的人