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.

519 lines
12 KiB

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