You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

555 lines
12 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
9 years ago
10 years ago
fix invalid memory address or nil pointer dereference error (Refs #762) https://github.com/tendermint/tendermint/issues/762#issuecomment-338276055 ``` E[10-19|04:52:38.969] Stopping peer for error module=p2p peer="Peer{MConn{178.62.46.14:46656} B14916FAF38A out}" err="Error: runtime error: invalid memory address or nil pointer dereference\nStack: goroutine 529485 [running]:\nruntime/debug.Stack(0xc4355cfb38, 0xb463e0, 0x11b1c30)\n\t/usr/local/go/src/runtime/debug/stack.go:24 +0xa7\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p.(*MConnection)._recover(0xc439a28870)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p/connection.go:206 +0x6e\npanic(0xb463e0, 0x11b1c30)\n\t/usr/local/go/src/runtime/panic.go:491 +0x283\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain.(*bpPeer).decrPending(0x0, 0x381)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain/pool.go:376 +0x22\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain.(*BlockPool).AddBlock(0xc4200e4000, 0xc4266d1f00, 0x40, 0xc432ac9640, 0x381)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain/pool.go:215 +0x139\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain.(*BlockchainReactor).Receive(0xc42050a780, 0xc420257740, 0x1171be0, 0xc42ff302d0, 0xc4384b2000, 0x381, 0x1000)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/blockchain/reactor.go:160 +0x712\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p.createMConnection.func1(0x11e5040, 0xc4384b2000, 0x381, 0x1000)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p/peer.go:334 +0xbd\ngithub.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p.(*MConnection).recvRoutine(0xc439a28870)\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p/connection.go:475 +0x4a3\ncreated by github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p.(*MConnection).OnStart\n\t/home/ubuntu/go/src/github.com/cosmos/gaia/vendor/github.com/tendermint/tendermint/p2p/connection.go:170 +0x187\n" ```
7 years ago
7 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package blockchain
  2. import (
  3. "math"
  4. "sync"
  5. "time"
  6. "github.com/tendermint/tendermint/types"
  7. cmn "github.com/tendermint/tmlibs/common"
  8. flow "github.com/tendermint/tmlibs/flowrate"
  9. "github.com/tendermint/tmlibs/log"
  10. )
  11. /*
  12. eg, L = latency = 0.1s
  13. P = num peers = 10
  14. FN = num full nodes
  15. BS = 1kB block size
  16. CB = 1 Mbit/s = 128 kB/s
  17. CB/P = 12.8 kB
  18. B/S = CB/P/BS = 12.8 blocks/s
  19. 12.8 * 0.1 = 1.28 blocks on conn
  20. */
  21. const (
  22. requestIntervalMS = 100
  23. maxTotalRequesters = 1000
  24. maxPendingRequests = maxTotalRequesters
  25. maxPendingRequestsPerPeer = 50
  26. minRecvRate = 10240 // 10Kb/s
  27. )
  28. var peerTimeoutSeconds = time.Duration(15) // not const so we can override with tests
  29. /*
  30. Peers self report their heights when we join the block pool.
  31. Starting from our latest pool.height, we request blocks
  32. in sequence from peers that reported higher heights than ours.
  33. Every so often we ask peers what height they're on so we can keep going.
  34. Requests are continuously made for blocks of higher heights until
  35. the limit is reached. If most of the requests have no available peers, and we
  36. are not at peer limits, we can probably switch to consensus reactor
  37. */
  38. type BlockPool struct {
  39. cmn.BaseService
  40. startTime time.Time
  41. mtx sync.Mutex
  42. // block requests
  43. requesters map[int64]*bpRequester
  44. height int64 // the lowest key in requesters.
  45. numPending int32 // number of requests pending assignment or block response
  46. // peers
  47. peers map[string]*bpPeer
  48. maxPeerHeight int64
  49. requestsCh chan<- BlockRequest
  50. timeoutsCh chan<- string
  51. }
  52. func NewBlockPool(start int64, requestsCh chan<- BlockRequest, timeoutsCh chan<- string) *BlockPool {
  53. bp := &BlockPool{
  54. peers: make(map[string]*bpPeer),
  55. requesters: make(map[int64]*bpRequester),
  56. height: start,
  57. numPending: 0,
  58. requestsCh: requestsCh,
  59. timeoutsCh: timeoutsCh,
  60. }
  61. bp.BaseService = *cmn.NewBaseService(nil, "BlockPool", bp)
  62. return bp
  63. }
  64. func (pool *BlockPool) OnStart() error {
  65. go pool.makeRequestersRoutine()
  66. pool.startTime = time.Now()
  67. return nil
  68. }
  69. func (pool *BlockPool) OnStop() {}
  70. // Run spawns requesters as needed.
  71. func (pool *BlockPool) makeRequestersRoutine() {
  72. for {
  73. if !pool.IsRunning() {
  74. break
  75. }
  76. _, numPending, lenRequesters := pool.GetStatus()
  77. if numPending >= maxPendingRequests {
  78. // sleep for a bit.
  79. time.Sleep(requestIntervalMS * time.Millisecond)
  80. // check for timed out peers
  81. pool.removeTimedoutPeers()
  82. } else if lenRequesters >= maxTotalRequesters {
  83. // sleep for a bit.
  84. time.Sleep(requestIntervalMS * time.Millisecond)
  85. // check for timed out peers
  86. pool.removeTimedoutPeers()
  87. } else {
  88. // request for more blocks.
  89. pool.makeNextRequester()
  90. }
  91. }
  92. }
  93. func (pool *BlockPool) removeTimedoutPeers() {
  94. pool.mtx.Lock()
  95. defer pool.mtx.Unlock()
  96. for _, peer := range pool.peers {
  97. if !peer.didTimeout && peer.numPending > 0 {
  98. curRate := peer.recvMonitor.Status().CurRate
  99. // XXX remove curRate != 0
  100. if curRate != 0 && curRate < minRecvRate {
  101. pool.sendTimeout(peer.id)
  102. pool.Logger.Error("SendTimeout", "peer", peer.id, "reason", "curRate too low")
  103. peer.didTimeout = true
  104. }
  105. }
  106. if peer.didTimeout {
  107. pool.removePeer(peer.id)
  108. }
  109. }
  110. }
  111. func (pool *BlockPool) GetStatus() (height int64, numPending int32, lenRequesters int) {
  112. pool.mtx.Lock()
  113. defer pool.mtx.Unlock()
  114. return pool.height, pool.numPending, len(pool.requesters)
  115. }
  116. // TODO: relax conditions, prevent abuse.
  117. func (pool *BlockPool) IsCaughtUp() bool {
  118. pool.mtx.Lock()
  119. defer pool.mtx.Unlock()
  120. // Need at least 1 peer to be considered caught up.
  121. if len(pool.peers) == 0 {
  122. pool.Logger.Debug("Blockpool has no peers")
  123. return false
  124. }
  125. // some conditions to determine if we're caught up
  126. receivedBlockOrTimedOut := (pool.height > 0 || time.Since(pool.startTime) > 5*time.Second)
  127. ourChainIsLongestAmongPeers := pool.maxPeerHeight == 0 || pool.height >= pool.maxPeerHeight
  128. isCaughtUp := receivedBlockOrTimedOut && ourChainIsLongestAmongPeers
  129. return isCaughtUp
  130. }
  131. // We need to see the second block's Commit to validate the first block.
  132. // So we peek two blocks at a time.
  133. // The caller will verify the commit.
  134. func (pool *BlockPool) PeekTwoBlocks() (first *types.Block, second *types.Block) {
  135. pool.mtx.Lock()
  136. defer pool.mtx.Unlock()
  137. if r := pool.requesters[pool.height]; r != nil {
  138. first = r.getBlock()
  139. }
  140. if r := pool.requesters[pool.height+1]; r != nil {
  141. second = r.getBlock()
  142. }
  143. return
  144. }
  145. // Pop the first block at pool.height
  146. // It must have been validated by 'second'.Commit from PeekTwoBlocks().
  147. func (pool *BlockPool) PopRequest() {
  148. pool.mtx.Lock()
  149. defer pool.mtx.Unlock()
  150. if r := pool.requesters[pool.height]; r != nil {
  151. /* The block can disappear at any time, due to removePeer().
  152. if r := pool.requesters[pool.height]; r == nil || r.block == nil {
  153. PanicSanity("PopRequest() requires a valid block")
  154. }
  155. */
  156. r.Stop()
  157. delete(pool.requesters, pool.height)
  158. pool.height++
  159. } else {
  160. cmn.PanicSanity(cmn.Fmt("Expected requester to pop, got nothing at height %v", pool.height))
  161. }
  162. }
  163. // Invalidates the block at pool.height,
  164. // Remove the peer and redo request from others.
  165. func (pool *BlockPool) RedoRequest(height int64) {
  166. pool.mtx.Lock()
  167. defer pool.mtx.Unlock()
  168. request := pool.requesters[height]
  169. if request.block == nil {
  170. cmn.PanicSanity("Expected block to be non-nil")
  171. }
  172. // RemovePeer will redo all requesters associated with this peer.
  173. // TODO: record this malfeasance
  174. pool.removePeer(request.peerID)
  175. }
  176. // TODO: ensure that blocks come in order for each peer.
  177. func (pool *BlockPool) AddBlock(peerID string, block *types.Block, blockSize int) {
  178. pool.mtx.Lock()
  179. defer pool.mtx.Unlock()
  180. requester := pool.requesters[block.Height]
  181. if requester == nil {
  182. // a block we didn't expect.
  183. // TODO:if height is too far ahead, punish peer
  184. return
  185. }
  186. if requester.setBlock(block, peerID) {
  187. pool.numPending--
  188. peer := pool.peers[peerID]
  189. if peer != nil {
  190. peer.decrPending(blockSize)
  191. }
  192. } else {
  193. // Bad peer?
  194. }
  195. }
  196. // MaxPeerHeight returns the highest height reported by a peer.
  197. func (pool *BlockPool) MaxPeerHeight() int64 {
  198. pool.mtx.Lock()
  199. defer pool.mtx.Unlock()
  200. return pool.maxPeerHeight
  201. }
  202. // Sets the peer's alleged blockchain height.
  203. func (pool *BlockPool) SetPeerHeight(peerID string, height int64) {
  204. pool.mtx.Lock()
  205. defer pool.mtx.Unlock()
  206. peer := pool.peers[peerID]
  207. if peer != nil {
  208. peer.height = height
  209. } else {
  210. peer = newBPPeer(pool, peerID, height)
  211. peer.setLogger(pool.Logger.With("peer", peerID))
  212. pool.peers[peerID] = peer
  213. }
  214. if height > pool.maxPeerHeight {
  215. pool.maxPeerHeight = height
  216. }
  217. }
  218. func (pool *BlockPool) RemovePeer(peerID string) {
  219. pool.mtx.Lock()
  220. defer pool.mtx.Unlock()
  221. pool.removePeer(peerID)
  222. }
  223. func (pool *BlockPool) removePeer(peerID string) {
  224. for _, requester := range pool.requesters {
  225. if requester.getPeerID() == peerID {
  226. if requester.getBlock() != nil {
  227. pool.numPending++
  228. }
  229. go requester.redo() // pick another peer and ...
  230. }
  231. }
  232. delete(pool.peers, peerID)
  233. }
  234. // Pick an available peer with at least the given minHeight.
  235. // If no peers are available, returns nil.
  236. func (pool *BlockPool) pickIncrAvailablePeer(minHeight int64) *bpPeer {
  237. pool.mtx.Lock()
  238. defer pool.mtx.Unlock()
  239. for _, peer := range pool.peers {
  240. if peer.didTimeout {
  241. pool.removePeer(peer.id)
  242. continue
  243. }
  244. if peer.numPending >= maxPendingRequestsPerPeer {
  245. continue
  246. }
  247. if peer.height < minHeight {
  248. continue
  249. }
  250. peer.incrPending()
  251. return peer
  252. }
  253. return nil
  254. }
  255. func (pool *BlockPool) makeNextRequester() {
  256. pool.mtx.Lock()
  257. defer pool.mtx.Unlock()
  258. nextHeight := pool.height + pool.requestersLen()
  259. request := newBPRequester(pool, nextHeight)
  260. // request.SetLogger(pool.Logger.With("height", nextHeight))
  261. pool.requesters[nextHeight] = request
  262. pool.numPending++
  263. err := request.Start()
  264. if err != nil {
  265. request.Logger.Error("Error starting request", "err", err)
  266. }
  267. }
  268. func (pool *BlockPool) requestersLen() int64 {
  269. return int64(len(pool.requesters))
  270. }
  271. func (pool *BlockPool) sendRequest(height int64, peerID string) {
  272. if !pool.IsRunning() {
  273. return
  274. }
  275. pool.requestsCh <- BlockRequest{height, peerID}
  276. }
  277. func (pool *BlockPool) sendTimeout(peerID string) {
  278. if !pool.IsRunning() {
  279. return
  280. }
  281. pool.timeoutsCh <- peerID
  282. }
  283. // unused by tendermint; left for debugging purposes
  284. func (pool *BlockPool) debug() string {
  285. pool.mtx.Lock() // Lock
  286. defer pool.mtx.Unlock()
  287. str := ""
  288. nextHeight := pool.height + pool.requestersLen()
  289. for h := pool.height; h < nextHeight; h++ {
  290. if pool.requesters[h] == nil {
  291. str += cmn.Fmt("H(%v):X ", h)
  292. } else {
  293. str += cmn.Fmt("H(%v):", h)
  294. str += cmn.Fmt("B?(%v) ", pool.requesters[h].block != nil)
  295. }
  296. }
  297. return str
  298. }
  299. //-------------------------------------
  300. type bpPeer struct {
  301. pool *BlockPool
  302. id string
  303. recvMonitor *flow.Monitor
  304. height int64
  305. numPending int32
  306. timeout *time.Timer
  307. didTimeout bool
  308. logger log.Logger
  309. }
  310. func newBPPeer(pool *BlockPool, peerID string, height int64) *bpPeer {
  311. peer := &bpPeer{
  312. pool: pool,
  313. id: peerID,
  314. height: height,
  315. numPending: 0,
  316. logger: log.NewNopLogger(),
  317. }
  318. return peer
  319. }
  320. func (peer *bpPeer) setLogger(l log.Logger) {
  321. peer.logger = l
  322. }
  323. func (peer *bpPeer) resetMonitor() {
  324. peer.recvMonitor = flow.New(time.Second, time.Second*40)
  325. initialValue := float64(minRecvRate) * math.E
  326. peer.recvMonitor.SetREMA(initialValue)
  327. }
  328. func (peer *bpPeer) resetTimeout() {
  329. if peer.timeout == nil {
  330. peer.timeout = time.AfterFunc(time.Second*peerTimeoutSeconds, peer.onTimeout)
  331. } else {
  332. peer.timeout.Reset(time.Second * peerTimeoutSeconds)
  333. }
  334. }
  335. func (peer *bpPeer) incrPending() {
  336. if peer.numPending == 0 {
  337. peer.resetMonitor()
  338. peer.resetTimeout()
  339. }
  340. peer.numPending++
  341. }
  342. func (peer *bpPeer) decrPending(recvSize int) {
  343. peer.numPending--
  344. if peer.numPending == 0 {
  345. peer.timeout.Stop()
  346. } else {
  347. peer.recvMonitor.Update(recvSize)
  348. peer.resetTimeout()
  349. }
  350. }
  351. func (peer *bpPeer) onTimeout() {
  352. peer.pool.mtx.Lock()
  353. defer peer.pool.mtx.Unlock()
  354. peer.pool.sendTimeout(peer.id)
  355. peer.logger.Error("SendTimeout", "reason", "onTimeout")
  356. peer.didTimeout = true
  357. }
  358. //-------------------------------------
  359. type bpRequester struct {
  360. cmn.BaseService
  361. pool *BlockPool
  362. height int64
  363. gotBlockCh chan struct{}
  364. redoCh chan struct{}
  365. mtx sync.Mutex
  366. peerID string
  367. block *types.Block
  368. }
  369. func newBPRequester(pool *BlockPool, height int64) *bpRequester {
  370. bpr := &bpRequester{
  371. pool: pool,
  372. height: height,
  373. gotBlockCh: make(chan struct{}),
  374. redoCh: make(chan struct{}),
  375. peerID: "",
  376. block: nil,
  377. }
  378. bpr.BaseService = *cmn.NewBaseService(nil, "bpRequester", bpr)
  379. return bpr
  380. }
  381. func (bpr *bpRequester) OnStart() error {
  382. go bpr.requestRoutine()
  383. return nil
  384. }
  385. // Returns true if the peer matches
  386. func (bpr *bpRequester) setBlock(block *types.Block, peerID string) bool {
  387. bpr.mtx.Lock()
  388. if bpr.block != nil || bpr.peerID != peerID {
  389. bpr.mtx.Unlock()
  390. return false
  391. }
  392. bpr.block = block
  393. bpr.mtx.Unlock()
  394. bpr.gotBlockCh <- struct{}{}
  395. return true
  396. }
  397. func (bpr *bpRequester) getBlock() *types.Block {
  398. bpr.mtx.Lock()
  399. defer bpr.mtx.Unlock()
  400. return bpr.block
  401. }
  402. func (bpr *bpRequester) getPeerID() string {
  403. bpr.mtx.Lock()
  404. defer bpr.mtx.Unlock()
  405. return bpr.peerID
  406. }
  407. func (bpr *bpRequester) reset() {
  408. bpr.mtx.Lock()
  409. bpr.peerID = ""
  410. bpr.block = nil
  411. bpr.mtx.Unlock()
  412. }
  413. // Tells bpRequester to pick another peer and try again.
  414. // NOTE: blocking
  415. func (bpr *bpRequester) redo() {
  416. bpr.redoCh <- struct{}{}
  417. }
  418. // Responsible for making more requests as necessary
  419. // Returns only when a block is found (e.g. AddBlock() is called)
  420. func (bpr *bpRequester) requestRoutine() {
  421. OUTER_LOOP:
  422. for {
  423. // Pick a peer to send request to.
  424. var peer *bpPeer = nil
  425. PICK_PEER_LOOP:
  426. for {
  427. if !bpr.IsRunning() || !bpr.pool.IsRunning() {
  428. return
  429. }
  430. peer = bpr.pool.pickIncrAvailablePeer(bpr.height)
  431. if peer == nil {
  432. //log.Info("No peers available", "height", height)
  433. time.Sleep(requestIntervalMS * time.Millisecond)
  434. continue PICK_PEER_LOOP
  435. }
  436. break PICK_PEER_LOOP
  437. }
  438. bpr.mtx.Lock()
  439. bpr.peerID = peer.id
  440. bpr.mtx.Unlock()
  441. // Send request and wait.
  442. bpr.pool.sendRequest(bpr.height, peer.id)
  443. select {
  444. case <-bpr.pool.Quit:
  445. bpr.Stop()
  446. return
  447. case <-bpr.Quit:
  448. return
  449. case <-bpr.redoCh:
  450. bpr.reset()
  451. continue OUTER_LOOP // When peer is removed
  452. case <-bpr.gotBlockCh:
  453. // We got the block, now see if it's good.
  454. select {
  455. case <-bpr.pool.Quit:
  456. bpr.Stop()
  457. return
  458. case <-bpr.Quit:
  459. return
  460. case <-bpr.redoCh:
  461. bpr.reset()
  462. continue OUTER_LOOP
  463. }
  464. }
  465. }
  466. }
  467. //-------------------------------------
  468. type BlockRequest struct {
  469. Height int64
  470. PeerID string
  471. }