fetcher.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905
  1. // Copyright 2016 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. package les
  17. import (
  18. "math/big"
  19. "sync"
  20. "time"
  21. "github.com/ethereum/go-ethereum/common"
  22. "github.com/ethereum/go-ethereum/common/mclock"
  23. "github.com/ethereum/go-ethereum/consensus"
  24. "github.com/ethereum/go-ethereum/core/rawdb"
  25. "github.com/ethereum/go-ethereum/core/types"
  26. "github.com/ethereum/go-ethereum/light"
  27. "github.com/ethereum/go-ethereum/log"
  28. )
  29. const (
  30. blockDelayTimeout = time.Second * 10 // timeout for a peer to announce a head that has already been confirmed by others
  31. maxNodeCount = 20 // maximum number of fetcherTreeNode entries remembered for each peer
  32. serverStateAvailable = 100 // number of recent blocks where state availability is assumed
  33. )
  34. // lightFetcher implements retrieval of newly announced headers. It also provides a peerHasBlock function for the
  35. // ODR system to ensure that we only request data related to a certain block from peers who have already processed
  36. // and announced that block.
  37. type lightFetcher struct {
  38. pm *ProtocolManager
  39. odr *LesOdr
  40. chain lightChain
  41. lock sync.Mutex // lock protects access to the fetcher's internal state variables except sent requests
  42. maxConfirmedTd *big.Int
  43. peers map[*peer]*fetcherPeerInfo
  44. lastUpdateStats *updateStatsEntry
  45. syncing bool
  46. syncDone chan *peer
  47. reqMu sync.RWMutex // reqMu protects access to sent header fetch requests
  48. requested map[uint64]fetchRequest
  49. deliverChn chan fetchResponse
  50. timeoutChn chan uint64
  51. requestTriggered bool
  52. requestTrigger chan struct{}
  53. lastTrustedHeader *types.Header
  54. }
  55. // lightChain extends the BlockChain interface by locking.
  56. type lightChain interface {
  57. BlockChain
  58. LockChain()
  59. UnlockChain()
  60. }
  61. // fetcherPeerInfo holds fetcher-specific information about each active peer
  62. type fetcherPeerInfo struct {
  63. root, lastAnnounced *fetcherTreeNode
  64. nodeCnt int
  65. confirmedTd *big.Int
  66. bestConfirmed *fetcherTreeNode
  67. nodeByHash map[common.Hash]*fetcherTreeNode
  68. firstUpdateStats *updateStatsEntry
  69. }
  70. // fetcherTreeNode is a node of a tree that holds information about blocks recently
  71. // announced and confirmed by a certain peer. Each new announce message from a peer
  72. // adds nodes to the tree, based on the previous announced head and the reorg depth.
  73. // There are three possible states for a tree node:
  74. // - announced: not downloaded (known) yet, but we know its head, number and td
  75. // - intermediate: not known, hash and td are empty, they are filled out when it becomes known
  76. // - known: both announced by this peer and downloaded (from any peer).
  77. // This structure makes it possible to always know which peer has a certain block,
  78. // which is necessary for selecting a suitable peer for ODR requests and also for
  79. // canonizing new heads. It also helps to always download the minimum necessary
  80. // amount of headers with a single request.
  81. type fetcherTreeNode struct {
  82. hash common.Hash
  83. number uint64
  84. td *big.Int
  85. known, requested bool
  86. parent *fetcherTreeNode
  87. children []*fetcherTreeNode
  88. }
  89. // fetchRequest represents a header download request
  90. type fetchRequest struct {
  91. hash common.Hash
  92. amount uint64
  93. peer *peer
  94. sent mclock.AbsTime
  95. timeout bool
  96. }
  97. // fetchResponse represents a header download response
  98. type fetchResponse struct {
  99. reqID uint64
  100. headers []*types.Header
  101. peer *peer
  102. }
  103. // newLightFetcher creates a new light fetcher
  104. func newLightFetcher(pm *ProtocolManager) *lightFetcher {
  105. f := &lightFetcher{
  106. pm: pm,
  107. chain: pm.blockchain.(*light.LightChain),
  108. odr: pm.odr,
  109. peers: make(map[*peer]*fetcherPeerInfo),
  110. deliverChn: make(chan fetchResponse, 100),
  111. requested: make(map[uint64]fetchRequest),
  112. timeoutChn: make(chan uint64),
  113. requestTrigger: make(chan struct{}, 1),
  114. syncDone: make(chan *peer),
  115. maxConfirmedTd: big.NewInt(0),
  116. }
  117. pm.peers.notify(f)
  118. f.pm.wg.Add(1)
  119. go f.syncLoop()
  120. return f
  121. }
  122. // syncLoop is the main event loop of the light fetcher
  123. func (f *lightFetcher) syncLoop() {
  124. defer f.pm.wg.Done()
  125. for {
  126. select {
  127. case <-f.pm.quitSync:
  128. return
  129. // request loop keeps running until no further requests are necessary or possible
  130. case <-f.requestTrigger:
  131. f.lock.Lock()
  132. var (
  133. rq *distReq
  134. reqID uint64
  135. syncing bool
  136. )
  137. if !f.syncing {
  138. rq, reqID, syncing = f.nextRequest()
  139. }
  140. f.requestTriggered = rq != nil
  141. f.lock.Unlock()
  142. if rq != nil {
  143. if _, ok := <-f.pm.reqDist.queue(rq); ok {
  144. if syncing {
  145. f.lock.Lock()
  146. f.syncing = true
  147. f.lock.Unlock()
  148. } else {
  149. go func() {
  150. time.Sleep(softRequestTimeout)
  151. f.reqMu.Lock()
  152. req, ok := f.requested[reqID]
  153. if ok {
  154. req.timeout = true
  155. f.requested[reqID] = req
  156. }
  157. f.reqMu.Unlock()
  158. // keep starting new requests while possible
  159. f.requestTrigger <- struct{}{}
  160. }()
  161. }
  162. } else {
  163. f.requestTrigger <- struct{}{}
  164. }
  165. }
  166. case reqID := <-f.timeoutChn:
  167. f.reqMu.Lock()
  168. req, ok := f.requested[reqID]
  169. if ok {
  170. delete(f.requested, reqID)
  171. }
  172. f.reqMu.Unlock()
  173. if ok {
  174. f.pm.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), true)
  175. req.peer.Log().Debug("Fetching data timed out hard")
  176. go f.pm.removePeer(req.peer.id)
  177. }
  178. case resp := <-f.deliverChn:
  179. f.reqMu.Lock()
  180. req, ok := f.requested[resp.reqID]
  181. if ok && req.peer != resp.peer {
  182. ok = false
  183. }
  184. if ok {
  185. delete(f.requested, resp.reqID)
  186. }
  187. f.reqMu.Unlock()
  188. if ok {
  189. f.pm.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), req.timeout)
  190. }
  191. f.lock.Lock()
  192. if !ok || !(f.syncing || f.processResponse(req, resp)) {
  193. resp.peer.Log().Debug("Failed processing response")
  194. go f.pm.removePeer(resp.peer.id)
  195. }
  196. f.lock.Unlock()
  197. case p := <-f.syncDone:
  198. f.lock.Lock()
  199. p.Log().Debug("Done synchronising with peer")
  200. f.checkSyncedHeaders(p)
  201. f.syncing = false
  202. f.lock.Unlock()
  203. f.requestTrigger <- struct{}{} // f.requestTriggered is always true here
  204. }
  205. }
  206. }
  207. // registerPeer adds a new peer to the fetcher's peer set
  208. func (f *lightFetcher) registerPeer(p *peer) {
  209. p.lock.Lock()
  210. p.hasBlock = func(hash common.Hash, number uint64, hasState bool) bool {
  211. return f.peerHasBlock(p, hash, number, hasState)
  212. }
  213. p.lock.Unlock()
  214. f.lock.Lock()
  215. defer f.lock.Unlock()
  216. f.peers[p] = &fetcherPeerInfo{nodeByHash: make(map[common.Hash]*fetcherTreeNode)}
  217. }
  218. // unregisterPeer removes a new peer from the fetcher's peer set
  219. func (f *lightFetcher) unregisterPeer(p *peer) {
  220. p.lock.Lock()
  221. p.hasBlock = nil
  222. p.lock.Unlock()
  223. f.lock.Lock()
  224. defer f.lock.Unlock()
  225. // check for potential timed out block delay statistics
  226. f.checkUpdateStats(p, nil)
  227. delete(f.peers, p)
  228. }
  229. // announce processes a new announcement message received from a peer, adding new
  230. // nodes to the peer's block tree and removing old nodes if necessary
  231. func (f *lightFetcher) announce(p *peer, head *announceData) {
  232. f.lock.Lock()
  233. defer f.lock.Unlock()
  234. p.Log().Debug("Received new announcement", "number", head.Number, "hash", head.Hash, "reorg", head.ReorgDepth)
  235. fp := f.peers[p]
  236. if fp == nil {
  237. p.Log().Debug("Announcement from unknown peer")
  238. return
  239. }
  240. if fp.lastAnnounced != nil && head.Td.Cmp(fp.lastAnnounced.td) <= 0 {
  241. // announced tds should be strictly monotonic
  242. p.Log().Debug("Received non-monotonic td", "current", head.Td, "previous", fp.lastAnnounced.td)
  243. go f.pm.removePeer(p.id)
  244. return
  245. }
  246. n := fp.lastAnnounced
  247. for i := uint64(0); i < head.ReorgDepth; i++ {
  248. if n == nil {
  249. break
  250. }
  251. n = n.parent
  252. }
  253. // n is now the reorg common ancestor, add a new branch of nodes
  254. if n != nil && (head.Number >= n.number+maxNodeCount || head.Number <= n.number) {
  255. // if announced head block height is lower or same as n or too far from it to add
  256. // intermediate nodes then discard previous announcement info and trigger a resync
  257. n = nil
  258. fp.nodeCnt = 0
  259. fp.nodeByHash = make(map[common.Hash]*fetcherTreeNode)
  260. }
  261. // check if the node count is too high to add new nodes, discard oldest ones if necessary
  262. if n != nil {
  263. // n is now the reorg common ancestor, add a new branch of nodes
  264. // check if the node count is too high to add new nodes
  265. locked := false
  266. for uint64(fp.nodeCnt)+head.Number-n.number > maxNodeCount && fp.root != nil {
  267. if !locked {
  268. f.chain.LockChain()
  269. defer f.chain.UnlockChain()
  270. locked = true
  271. }
  272. // if one of root's children is canonical, keep it, delete other branches and root itself
  273. var newRoot *fetcherTreeNode
  274. for i, nn := range fp.root.children {
  275. if rawdb.ReadCanonicalHash(f.pm.chainDb, nn.number) == nn.hash {
  276. fp.root.children = append(fp.root.children[:i], fp.root.children[i+1:]...)
  277. nn.parent = nil
  278. newRoot = nn
  279. break
  280. }
  281. }
  282. fp.deleteNode(fp.root)
  283. if n == fp.root {
  284. n = newRoot
  285. }
  286. fp.root = newRoot
  287. if newRoot == nil || !f.checkKnownNode(p, newRoot) {
  288. fp.bestConfirmed = nil
  289. fp.confirmedTd = nil
  290. }
  291. if n == nil {
  292. break
  293. }
  294. }
  295. if n != nil {
  296. for n.number < head.Number {
  297. nn := &fetcherTreeNode{number: n.number + 1, parent: n}
  298. n.children = append(n.children, nn)
  299. n = nn
  300. fp.nodeCnt++
  301. }
  302. n.hash = head.Hash
  303. n.td = head.Td
  304. fp.nodeByHash[n.hash] = n
  305. }
  306. }
  307. if n == nil {
  308. // could not find reorg common ancestor or had to delete entire tree, a new root and a resync is needed
  309. if fp.root != nil {
  310. fp.deleteNode(fp.root)
  311. }
  312. n = &fetcherTreeNode{hash: head.Hash, number: head.Number, td: head.Td}
  313. fp.root = n
  314. fp.nodeCnt++
  315. fp.nodeByHash[n.hash] = n
  316. fp.bestConfirmed = nil
  317. fp.confirmedTd = nil
  318. }
  319. f.checkKnownNode(p, n)
  320. p.lock.Lock()
  321. p.headInfo = head
  322. fp.lastAnnounced = n
  323. p.lock.Unlock()
  324. f.checkUpdateStats(p, nil)
  325. if !f.requestTriggered {
  326. f.requestTriggered = true
  327. f.requestTrigger <- struct{}{}
  328. }
  329. }
  330. // peerHasBlock returns true if we can assume the peer knows the given block
  331. // based on its announcements
  332. func (f *lightFetcher) peerHasBlock(p *peer, hash common.Hash, number uint64, hasState bool) bool {
  333. f.lock.Lock()
  334. defer f.lock.Unlock()
  335. fp := f.peers[p]
  336. if fp == nil || fp.root == nil {
  337. return false
  338. }
  339. if hasState {
  340. if fp.lastAnnounced == nil || fp.lastAnnounced.number > number+serverStateAvailable {
  341. return false
  342. }
  343. }
  344. if f.syncing {
  345. // always return true when syncing
  346. // false positives are acceptable, a more sophisticated condition can be implemented later
  347. return true
  348. }
  349. if number >= fp.root.number {
  350. // it is recent enough that if it is known, is should be in the peer's block tree
  351. return fp.nodeByHash[hash] != nil
  352. }
  353. f.chain.LockChain()
  354. defer f.chain.UnlockChain()
  355. // if it's older than the peer's block tree root but it's in the same canonical chain
  356. // as the root, we can still be sure the peer knows it
  357. //
  358. // when syncing, just check if it is part of the known chain, there is nothing better we
  359. // can do since we do not know the most recent block hash yet
  360. return rawdb.ReadCanonicalHash(f.pm.chainDb, fp.root.number) == fp.root.hash && rawdb.ReadCanonicalHash(f.pm.chainDb, number) == hash
  361. }
  362. // requestAmount calculates the amount of headers to be downloaded starting
  363. // from a certain head backwards
  364. func (f *lightFetcher) requestAmount(p *peer, n *fetcherTreeNode) uint64 {
  365. amount := uint64(0)
  366. nn := n
  367. for nn != nil && !f.checkKnownNode(p, nn) {
  368. nn = nn.parent
  369. amount++
  370. }
  371. if nn == nil {
  372. amount = n.number
  373. }
  374. return amount
  375. }
  376. // requestedID tells if a certain reqID has been requested by the fetcher
  377. func (f *lightFetcher) requestedID(reqID uint64) bool {
  378. f.reqMu.RLock()
  379. _, ok := f.requested[reqID]
  380. f.reqMu.RUnlock()
  381. return ok
  382. }
  383. // nextRequest selects the peer and announced head to be requested next, amount
  384. // to be downloaded starting from the head backwards is also returned
  385. func (f *lightFetcher) nextRequest() (*distReq, uint64, bool) {
  386. var (
  387. bestHash common.Hash
  388. bestAmount uint64
  389. bestTd *big.Int
  390. bestSyncing bool
  391. )
  392. bestHash, bestAmount, bestTd, bestSyncing = f.findBestRequest()
  393. if bestTd == f.maxConfirmedTd {
  394. return nil, 0, false
  395. }
  396. var rq *distReq
  397. reqID := genReqID()
  398. if bestSyncing {
  399. rq = f.newFetcherDistReqForSync(bestHash)
  400. } else {
  401. rq = f.newFetcherDistReq(bestHash, reqID, bestAmount)
  402. }
  403. return rq, reqID, bestSyncing
  404. }
  405. // findBestRequest finds the best head to request that has been announced by but not yet requested from a known peer.
  406. // It also returns the announced Td (which should be verified after fetching the head),
  407. // the necessary amount to request and whether a downloader sync is necessary instead of a normal header request.
  408. func (f *lightFetcher) findBestRequest() (bestHash common.Hash, bestAmount uint64, bestTd *big.Int, bestSyncing bool) {
  409. bestTd = f.maxConfirmedTd
  410. bestSyncing = false
  411. for p, fp := range f.peers {
  412. for hash, n := range fp.nodeByHash {
  413. if f.checkKnownNode(p, n) || n.requested {
  414. continue
  415. }
  416. //if ulc mode is disabled, isTrustedHash returns true
  417. amount := f.requestAmount(p, n)
  418. if (bestTd == nil || n.td.Cmp(bestTd) > 0 || amount < bestAmount) && (f.isTrustedHash(hash) || f.maxConfirmedTd.Int64() == 0) {
  419. bestHash = hash
  420. bestTd = n.td
  421. bestAmount = amount
  422. bestSyncing = fp.bestConfirmed == nil || fp.root == nil || !f.checkKnownNode(p, fp.root)
  423. }
  424. }
  425. }
  426. return
  427. }
  428. // isTrustedHash checks if the block can be trusted by the minimum trusted fraction.
  429. func (f *lightFetcher) isTrustedHash(hash common.Hash) bool {
  430. if !f.pm.isULCEnabled() {
  431. return true
  432. }
  433. var numAgreed int
  434. for p, fp := range f.peers {
  435. if !p.isTrusted {
  436. continue
  437. }
  438. if _, ok := fp.nodeByHash[hash]; !ok {
  439. continue
  440. }
  441. numAgreed++
  442. }
  443. return 100*numAgreed/len(f.pm.ulc.trustedKeys) >= f.pm.ulc.minTrustedFraction
  444. }
  445. func (f *lightFetcher) newFetcherDistReqForSync(bestHash common.Hash) *distReq {
  446. return &distReq{
  447. getCost: func(dp distPeer) uint64 {
  448. return 0
  449. },
  450. canSend: func(dp distPeer) bool {
  451. p := dp.(*peer)
  452. f.lock.Lock()
  453. defer f.lock.Unlock()
  454. if p.isOnlyAnnounce {
  455. return false
  456. }
  457. fp := f.peers[p]
  458. return fp != nil && fp.nodeByHash[bestHash] != nil
  459. },
  460. request: func(dp distPeer) func() {
  461. if f.pm.isULCEnabled() {
  462. //keep last trusted header before sync
  463. f.setLastTrustedHeader(f.chain.CurrentHeader())
  464. }
  465. go func() {
  466. p := dp.(*peer)
  467. p.Log().Debug("Synchronisation started")
  468. f.pm.synchronise(p)
  469. f.syncDone <- p
  470. }()
  471. return nil
  472. },
  473. }
  474. }
  475. // newFetcherDistReq creates a new request for the distributor.
  476. func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bestAmount uint64) *distReq {
  477. return &distReq{
  478. getCost: func(dp distPeer) uint64 {
  479. p := dp.(*peer)
  480. return p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
  481. },
  482. canSend: func(dp distPeer) bool {
  483. p := dp.(*peer)
  484. f.lock.Lock()
  485. defer f.lock.Unlock()
  486. if p.isOnlyAnnounce {
  487. return false
  488. }
  489. fp := f.peers[p]
  490. if fp == nil {
  491. return false
  492. }
  493. n := fp.nodeByHash[bestHash]
  494. return n != nil && !n.requested
  495. },
  496. request: func(dp distPeer) func() {
  497. p := dp.(*peer)
  498. f.lock.Lock()
  499. fp := f.peers[p]
  500. if fp != nil {
  501. n := fp.nodeByHash[bestHash]
  502. if n != nil {
  503. n.requested = true
  504. }
  505. }
  506. f.lock.Unlock()
  507. cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
  508. p.fcServer.QueuedRequest(reqID, cost)
  509. f.reqMu.Lock()
  510. f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}
  511. f.reqMu.Unlock()
  512. go func() {
  513. time.Sleep(hardRequestTimeout)
  514. f.timeoutChn <- reqID
  515. }()
  516. return func() { p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true) }
  517. },
  518. }
  519. }
  520. // deliverHeaders delivers header download request responses for processing
  521. func (f *lightFetcher) deliverHeaders(peer *peer, reqID uint64, headers []*types.Header) {
  522. f.deliverChn <- fetchResponse{reqID: reqID, headers: headers, peer: peer}
  523. }
  524. // processResponse processes header download request responses, returns true if successful
  525. func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) bool {
  526. if uint64(len(resp.headers)) != req.amount || resp.headers[0].Hash() != req.hash {
  527. req.peer.Log().Debug("Response content mismatch", "requested", len(resp.headers), "reqfrom", resp.headers[0], "delivered", req.amount, "delfrom", req.hash)
  528. return false
  529. }
  530. headers := make([]*types.Header, req.amount)
  531. for i, header := range resp.headers {
  532. headers[int(req.amount)-1-i] = header
  533. }
  534. if _, err := f.chain.InsertHeaderChain(headers, 1); err != nil {
  535. if err == consensus.ErrFutureBlock {
  536. return true
  537. }
  538. log.Debug("Failed to insert header chain", "err", err)
  539. return false
  540. }
  541. tds := make([]*big.Int, len(headers))
  542. for i, header := range headers {
  543. td := f.chain.GetTd(header.Hash(), header.Number.Uint64())
  544. if td == nil {
  545. log.Debug("Total difficulty not found for header", "index", i+1, "number", header.Number, "hash", header.Hash())
  546. return false
  547. }
  548. tds[i] = td
  549. }
  550. f.newHeaders(headers, tds)
  551. return true
  552. }
  553. // newHeaders updates the block trees of all active peers according to a newly
  554. // downloaded and validated batch or headers
  555. func (f *lightFetcher) newHeaders(headers []*types.Header, tds []*big.Int) {
  556. var maxTd *big.Int
  557. for p, fp := range f.peers {
  558. if !f.checkAnnouncedHeaders(fp, headers, tds) {
  559. p.Log().Debug("Inconsistent announcement")
  560. go f.pm.removePeer(p.id)
  561. }
  562. if fp.confirmedTd != nil && (maxTd == nil || maxTd.Cmp(fp.confirmedTd) > 0) {
  563. maxTd = fp.confirmedTd
  564. }
  565. }
  566. if maxTd != nil {
  567. f.updateMaxConfirmedTd(maxTd)
  568. }
  569. }
  570. // checkAnnouncedHeaders updates peer's block tree if necessary after validating
  571. // a batch of headers. It searches for the latest header in the batch that has a
  572. // matching tree node (if any), and if it has not been marked as known already,
  573. // sets it and its parents to known (even those which are older than the currently
  574. // validated ones). Return value shows if all hashes, numbers and Tds matched
  575. // correctly to the announced values (otherwise the peer should be dropped).
  576. func (f *lightFetcher) checkAnnouncedHeaders(fp *fetcherPeerInfo, headers []*types.Header, tds []*big.Int) bool {
  577. var (
  578. n *fetcherTreeNode
  579. header *types.Header
  580. td *big.Int
  581. )
  582. for i := len(headers) - 1; ; i-- {
  583. if i < 0 {
  584. if n == nil {
  585. // no more headers and nothing to match
  586. return true
  587. }
  588. // we ran out of recently delivered headers but have not reached a node known by this peer yet, continue matching
  589. hash, number := header.ParentHash, header.Number.Uint64()-1
  590. td = f.chain.GetTd(hash, number)
  591. header = f.chain.GetHeader(hash, number)
  592. if header == nil || td == nil {
  593. log.Error("Missing parent of validated header", "hash", hash, "number", number)
  594. return false
  595. }
  596. } else {
  597. header = headers[i]
  598. td = tds[i]
  599. }
  600. hash := header.Hash()
  601. number := header.Number.Uint64()
  602. if n == nil {
  603. n = fp.nodeByHash[hash]
  604. }
  605. if n != nil {
  606. if n.td == nil {
  607. // node was unannounced
  608. if nn := fp.nodeByHash[hash]; nn != nil {
  609. // if there was already a node with the same hash, continue there and drop this one
  610. nn.children = append(nn.children, n.children...)
  611. n.children = nil
  612. fp.deleteNode(n)
  613. n = nn
  614. } else {
  615. n.hash = hash
  616. n.td = td
  617. fp.nodeByHash[hash] = n
  618. }
  619. }
  620. // check if it matches the header
  621. if n.hash != hash || n.number != number || n.td.Cmp(td) != 0 {
  622. // peer has previously made an invalid announcement
  623. return false
  624. }
  625. if n.known {
  626. // we reached a known node that matched our expectations, return with success
  627. return true
  628. }
  629. n.known = true
  630. if fp.confirmedTd == nil || td.Cmp(fp.confirmedTd) > 0 {
  631. fp.confirmedTd = td
  632. fp.bestConfirmed = n
  633. }
  634. n = n.parent
  635. if n == nil {
  636. return true
  637. }
  638. }
  639. }
  640. }
  641. // checkSyncedHeaders updates peer's block tree after synchronisation by marking
  642. // downloaded headers as known. If none of the announced headers are found after
  643. // syncing, the peer is dropped.
  644. func (f *lightFetcher) checkSyncedHeaders(p *peer) {
  645. fp := f.peers[p]
  646. if fp == nil {
  647. p.Log().Debug("Unknown peer to check sync headers")
  648. return
  649. }
  650. n := fp.lastAnnounced
  651. var td *big.Int
  652. var h *types.Header
  653. if f.pm.isULCEnabled() {
  654. var unapprovedHashes []common.Hash
  655. // Overwrite last announced for ULC mode
  656. h, unapprovedHashes = f.lastTrustedTreeNode(p)
  657. //rollback untrusted blocks
  658. f.chain.Rollback(unapprovedHashes)
  659. //overwrite to last trusted
  660. n = fp.nodeByHash[h.Hash()]
  661. }
  662. //find last valid block
  663. for n != nil {
  664. if td = f.chain.GetTd(n.hash, n.number); td != nil {
  665. break
  666. }
  667. n = n.parent
  668. }
  669. // Now n is the latest downloaded/approved header after syncing
  670. if n == nil {
  671. p.Log().Debug("Synchronisation failed")
  672. go f.pm.removePeer(p.id)
  673. return
  674. }
  675. header := f.chain.GetHeader(n.hash, n.number)
  676. f.newHeaders([]*types.Header{header}, []*big.Int{td})
  677. }
  678. // lastTrustedTreeNode return last approved treeNode and a list of unapproved hashes
  679. func (f *lightFetcher) lastTrustedTreeNode(p *peer) (*types.Header, []common.Hash) {
  680. unapprovedHashes := make([]common.Hash, 0)
  681. current := f.chain.CurrentHeader()
  682. if f.lastTrustedHeader == nil {
  683. return current, unapprovedHashes
  684. }
  685. canonical := f.chain.CurrentHeader()
  686. if canonical.Number.Uint64() > f.lastTrustedHeader.Number.Uint64() {
  687. canonical = f.chain.GetHeaderByNumber(f.lastTrustedHeader.Number.Uint64())
  688. }
  689. commonAncestor := rawdb.FindCommonAncestor(f.pm.chainDb, canonical, f.lastTrustedHeader)
  690. if commonAncestor == nil {
  691. log.Error("Common ancestor of last trusted header and canonical header is nil", "canonical hash", canonical.Hash(), "trusted hash", f.lastTrustedHeader.Hash())
  692. return current, unapprovedHashes
  693. }
  694. for current.Hash() == commonAncestor.Hash() {
  695. if f.isTrustedHash(current.Hash()) {
  696. break
  697. }
  698. unapprovedHashes = append(unapprovedHashes, current.Hash())
  699. current = f.chain.GetHeader(current.ParentHash, current.Number.Uint64()-1)
  700. }
  701. return current, unapprovedHashes
  702. }
  703. func (f *lightFetcher) setLastTrustedHeader(h *types.Header) {
  704. f.lock.Lock()
  705. defer f.lock.Unlock()
  706. f.lastTrustedHeader = h
  707. }
  708. // checkKnownNode checks if a block tree node is known (downloaded and validated)
  709. // If it was not known previously but found in the database, sets its known flag
  710. func (f *lightFetcher) checkKnownNode(p *peer, n *fetcherTreeNode) bool {
  711. if n.known {
  712. return true
  713. }
  714. td := f.chain.GetTd(n.hash, n.number)
  715. if td == nil {
  716. return false
  717. }
  718. header := f.chain.GetHeader(n.hash, n.number)
  719. // check the availability of both header and td because reads are not protected by chain db mutex
  720. // Note: returning false is always safe here
  721. if header == nil {
  722. return false
  723. }
  724. fp := f.peers[p]
  725. if fp == nil {
  726. p.Log().Debug("Unknown peer to check known nodes")
  727. return false
  728. }
  729. if !f.checkAnnouncedHeaders(fp, []*types.Header{header}, []*big.Int{td}) {
  730. p.Log().Debug("Inconsistent announcement")
  731. go f.pm.removePeer(p.id)
  732. }
  733. if fp.confirmedTd != nil {
  734. f.updateMaxConfirmedTd(fp.confirmedTd)
  735. }
  736. return n.known
  737. }
  738. // deleteNode deletes a node and its child subtrees from a peer's block tree
  739. func (fp *fetcherPeerInfo) deleteNode(n *fetcherTreeNode) {
  740. if n.parent != nil {
  741. for i, nn := range n.parent.children {
  742. if nn == n {
  743. n.parent.children = append(n.parent.children[:i], n.parent.children[i+1:]...)
  744. break
  745. }
  746. }
  747. }
  748. for {
  749. if n.td != nil {
  750. delete(fp.nodeByHash, n.hash)
  751. }
  752. fp.nodeCnt--
  753. if len(n.children) == 0 {
  754. return
  755. }
  756. for i, nn := range n.children {
  757. if i == 0 {
  758. n = nn
  759. } else {
  760. fp.deleteNode(nn)
  761. }
  762. }
  763. }
  764. }
  765. // updateStatsEntry items form a linked list that is expanded with a new item every time a new head with a higher Td
  766. // than the previous one has been downloaded and validated. The list contains a series of maximum confirmed Td values
  767. // and the time these values have been confirmed, both increasing monotonically. A maximum confirmed Td is calculated
  768. // both globally for all peers and also for each individual peer (meaning that the given peer has announced the head
  769. // and it has also been downloaded from any peer, either before or after the given announcement).
  770. // The linked list has a global tail where new confirmed Td entries are added and a separate head for each peer,
  771. // pointing to the next Td entry that is higher than the peer's max confirmed Td (nil if it has already confirmed
  772. // the current global head).
  773. type updateStatsEntry struct {
  774. time mclock.AbsTime
  775. td *big.Int
  776. next *updateStatsEntry
  777. }
  778. // updateMaxConfirmedTd updates the block delay statistics of active peers. Whenever a new highest Td is confirmed,
  779. // adds it to the end of a linked list together with the time it has been confirmed. Then checks which peers have
  780. // already confirmed a head with the same or higher Td (which counts as zero block delay) and updates their statistics.
  781. // Those who have not confirmed such a head by now will be updated by a subsequent checkUpdateStats call with a
  782. // positive block delay value.
  783. func (f *lightFetcher) updateMaxConfirmedTd(td *big.Int) {
  784. if f.maxConfirmedTd == nil || td.Cmp(f.maxConfirmedTd) > 0 {
  785. f.maxConfirmedTd = td
  786. newEntry := &updateStatsEntry{
  787. time: mclock.Now(),
  788. td: td,
  789. }
  790. if f.lastUpdateStats != nil {
  791. f.lastUpdateStats.next = newEntry
  792. }
  793. f.lastUpdateStats = newEntry
  794. for p := range f.peers {
  795. f.checkUpdateStats(p, newEntry)
  796. }
  797. }
  798. }
  799. // checkUpdateStats checks those peers who have not confirmed a certain highest Td (or a larger one) by the time it
  800. // has been confirmed by another peer. If they have confirmed such a head by now, their stats are updated with the
  801. // block delay which is (this peer's confirmation time)-(first confirmation time). After blockDelayTimeout has passed,
  802. // the stats are updated with blockDelayTimeout value. In either case, the confirmed or timed out updateStatsEntry
  803. // items are removed from the head of the linked list.
  804. // If a new entry has been added to the global tail, it is passed as a parameter here even though this function
  805. // assumes that it has already been added, so that if the peer's list is empty (all heads confirmed, head is nil),
  806. // it can set the new head to newEntry.
  807. func (f *lightFetcher) checkUpdateStats(p *peer, newEntry *updateStatsEntry) {
  808. now := mclock.Now()
  809. fp := f.peers[p]
  810. if fp == nil {
  811. p.Log().Debug("Unknown peer to check update stats")
  812. return
  813. }
  814. if newEntry != nil && fp.firstUpdateStats == nil {
  815. fp.firstUpdateStats = newEntry
  816. }
  817. for fp.firstUpdateStats != nil && fp.firstUpdateStats.time <= now-mclock.AbsTime(blockDelayTimeout) {
  818. f.pm.serverPool.adjustBlockDelay(p.poolEntry, blockDelayTimeout)
  819. fp.firstUpdateStats = fp.firstUpdateStats.next
  820. }
  821. if fp.confirmedTd != nil {
  822. for fp.firstUpdateStats != nil && fp.firstUpdateStats.td.Cmp(fp.confirmedTd) <= 0 {
  823. f.pm.serverPool.adjustBlockDelay(p.poolEntry, time.Duration(now-fp.firstUpdateStats.time))
  824. fp.firstUpdateStats = fp.firstUpdateStats.next
  825. }
  826. }
  827. }