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.

479 lines
14 KiB

blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
8 years ago
7 years ago
7 years ago
7 years ago
8 years ago
10 years ago
8 years ago
10 years ago
blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
10 years ago
blockchain: Reorg reactor (#3561) * go routines in blockchain reactor * Added reference to the go routine diagram * Initial commit * cleanup * Undo testing_logger change, committed by mistake * Fix the test loggers * pulled some fsm code into pool.go * added pool tests * changes to the design added block requests under peer moved the request trigger in the reactor poolRoutine, triggered now by a ticker in general moved everything required for making block requests smarter in the poolRoutine added a simple map of heights to keep track of what will need to be requested next added a few more tests * send errors to FSM in a different channel than blocks send errors (RemovePeer) from switch on a different channel than the one receiving blocks renamed channels added more pool tests * more pool tests * lint errors * more tests * more tests * switch fast sync to new implementation * fixed data race in tests * cleanup * finished fsm tests * address golangci comments :) * address golangci comments :) * Added timeout on next block needed to advance * updating docs and cleanup * fix issue in test from previous cleanup * cleanup * Added termination scenarios, tests and more cleanup * small fixes to adr, comments and cleanup * Fix bug in sendRequest() If we tried to send a request to a peer not present in the switch, a missing continue statement caused the request to be blackholed in a peer that was removed and never retried. While this bug was manifesting, the reactor kept asking for other blocks that would be stored and never consumed. Added the number of unconsumed blocks in the math for requesting blocks ahead of current processing height so eventually there will be no more blocks requested until the already received ones are consumed. * remove bpPeer's didTimeout field * Use distinct err codes for peer timeout and FSM timeouts * Don't allow peers to update with lower height * review comments from Ethan and Zarko * some cleanup, renaming, comments * Move block execution in separate goroutine * Remove pool's numPending * review comments * fix lint, remove old blockchain reactor and duplicates in fsm tests * small reorg around peer after review comments * add the reactor spec * verify block only once * review comments * change to int for max number of pending requests * cleanup and godoc * Add configuration flag fast sync version * golangci fixes * fix config template * move both reactor versions under blockchain * cleanup, golint, renaming stuff * updated documentation, fixed more golint warnings * integrate with behavior package * sync with master * gofmt * add changelog_pending entry * move to improvments * suggestion to changelog entry
5 years ago
  1. package v0
  2. import (
  3. "errors"
  4. "fmt"
  5. "reflect"
  6. "time"
  7. amino "github.com/tendermint/go-amino"
  8. "github.com/tendermint/tendermint/libs/log"
  9. "github.com/tendermint/tendermint/p2p"
  10. sm "github.com/tendermint/tendermint/state"
  11. "github.com/tendermint/tendermint/store"
  12. "github.com/tendermint/tendermint/types"
  13. )
  14. const (
  15. // BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
  16. BlockchainChannel = byte(0x40)
  17. trySyncIntervalMS = 10
  18. // stop syncing when last block's time is
  19. // within this much of the system time.
  20. // stopSyncingDurationMinutes = 10
  21. // ask for best height every 10s
  22. statusUpdateIntervalSeconds = 10
  23. // check if we should switch to consensus reactor
  24. switchToConsensusIntervalSeconds = 1
  25. // NOTE: keep up to date with bcBlockResponseMessage
  26. bcBlockResponseMessagePrefixSize = 4
  27. bcBlockResponseMessageFieldKeySize = 1
  28. maxMsgSize = types.MaxBlockSizeBytes +
  29. bcBlockResponseMessagePrefixSize +
  30. bcBlockResponseMessageFieldKeySize
  31. )
  32. type consensusReactor interface {
  33. // for when we switch from blockchain reactor and fast sync to
  34. // the consensus machine
  35. SwitchToConsensus(sm.State, int)
  36. }
  37. type peerError struct {
  38. err error
  39. peerID p2p.ID
  40. }
  41. func (e peerError) Error() string {
  42. return fmt.Sprintf("error with peer %v: %s", e.peerID, e.err.Error())
  43. }
  44. // BlockchainReactor handles long-term catchup syncing.
  45. type BlockchainReactor struct {
  46. p2p.BaseReactor
  47. // immutable
  48. initialState sm.State
  49. blockExec *sm.BlockExecutor
  50. store *store.BlockStore
  51. pool *BlockPool
  52. fastSync bool
  53. requestsCh <-chan BlockRequest
  54. errorsCh <-chan peerError
  55. }
  56. // NewBlockchainReactor returns new reactor instance.
  57. func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *store.BlockStore,
  58. fastSync bool) *BlockchainReactor {
  59. if state.LastBlockHeight != store.Height() {
  60. panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight,
  61. store.Height()))
  62. }
  63. requestsCh := make(chan BlockRequest, maxTotalRequesters)
  64. const capacity = 1000 // must be bigger than peers count
  65. errorsCh := make(chan peerError, capacity) // so we don't block in #Receive#pool.AddBlock
  66. pool := NewBlockPool(
  67. store.Height()+1,
  68. requestsCh,
  69. errorsCh,
  70. )
  71. bcR := &BlockchainReactor{
  72. initialState: state,
  73. blockExec: blockExec,
  74. store: store,
  75. pool: pool,
  76. fastSync: fastSync,
  77. requestsCh: requestsCh,
  78. errorsCh: errorsCh,
  79. }
  80. bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
  81. return bcR
  82. }
  83. // SetLogger implements cmn.Service by setting the logger on reactor and pool.
  84. func (bcR *BlockchainReactor) SetLogger(l log.Logger) {
  85. bcR.BaseService.Logger = l
  86. bcR.pool.Logger = l
  87. }
  88. // OnStart implements cmn.Service.
  89. func (bcR *BlockchainReactor) OnStart() error {
  90. if bcR.fastSync {
  91. err := bcR.pool.Start()
  92. if err != nil {
  93. return err
  94. }
  95. go bcR.poolRoutine()
  96. }
  97. return nil
  98. }
  99. // OnStop implements cmn.Service.
  100. func (bcR *BlockchainReactor) OnStop() {
  101. bcR.pool.Stop()
  102. }
  103. // GetChannels implements Reactor
  104. func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
  105. return []*p2p.ChannelDescriptor{
  106. {
  107. ID: BlockchainChannel,
  108. Priority: 10,
  109. SendQueueCapacity: 1000,
  110. RecvBufferCapacity: 50 * 4096,
  111. RecvMessageCapacity: maxMsgSize,
  112. },
  113. }
  114. }
  115. // AddPeer implements Reactor by sending our state to peer.
  116. func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) {
  117. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusResponseMessage{bcR.store.Height()})
  118. peer.Send(BlockchainChannel, msgBytes)
  119. // it's OK if send fails. will try later in poolRoutine
  120. // peer is added to the pool once we receive the first
  121. // bcStatusResponseMessage from the peer and call pool.SetPeerHeight
  122. }
  123. // RemovePeer implements Reactor by removing peer from the pool.
  124. func (bcR *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
  125. bcR.pool.RemovePeer(peer.ID())
  126. }
  127. // respondToPeer loads a block and sends it to the requesting peer,
  128. // if we have it. Otherwise, we'll respond saying we don't have it.
  129. // According to the Tendermint spec, if all nodes are honest,
  130. // no node should be requesting for a block that's non-existent.
  131. func (bcR *BlockchainReactor) respondToPeer(msg *bcBlockRequestMessage,
  132. src p2p.Peer) (queued bool) {
  133. block := bcR.store.LoadBlock(msg.Height)
  134. if block != nil {
  135. msgBytes := cdc.MustMarshalBinaryBare(&bcBlockResponseMessage{Block: block})
  136. return src.TrySend(BlockchainChannel, msgBytes)
  137. }
  138. bcR.Logger.Info("Peer asking for a block we don't have", "src", src, "height", msg.Height)
  139. msgBytes := cdc.MustMarshalBinaryBare(&bcNoBlockResponseMessage{Height: msg.Height})
  140. return src.TrySend(BlockchainChannel, msgBytes)
  141. }
  142. // Receive implements Reactor by handling 4 types of messages (look below).
  143. func (bcR *BlockchainReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  144. msg, err := decodeMsg(msgBytes)
  145. if err != nil {
  146. bcR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
  147. bcR.Switch.StopPeerForError(src, err)
  148. return
  149. }
  150. if err = msg.ValidateBasic(); err != nil {
  151. bcR.Logger.Error("Peer sent us invalid msg", "peer", src, "msg", msg, "err", err)
  152. bcR.Switch.StopPeerForError(src, err)
  153. return
  154. }
  155. bcR.Logger.Debug("Receive", "src", src, "chID", chID, "msg", msg)
  156. switch msg := msg.(type) {
  157. case *bcBlockRequestMessage:
  158. bcR.respondToPeer(msg, src)
  159. case *bcBlockResponseMessage:
  160. bcR.pool.AddBlock(src.ID(), msg.Block, len(msgBytes))
  161. case *bcStatusRequestMessage:
  162. // Send peer our state.
  163. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusResponseMessage{bcR.store.Height()})
  164. src.TrySend(BlockchainChannel, msgBytes)
  165. case *bcStatusResponseMessage:
  166. // Got a peer status. Unverified.
  167. bcR.pool.SetPeerHeight(src.ID(), msg.Height)
  168. default:
  169. bcR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  170. }
  171. }
  172. // Handle messages from the poolReactor telling the reactor what to do.
  173. // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
  174. func (bcR *BlockchainReactor) poolRoutine() {
  175. trySyncTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
  176. statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
  177. switchToConsensusTicker := time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
  178. blocksSynced := 0
  179. chainID := bcR.initialState.ChainID
  180. state := bcR.initialState
  181. lastHundred := time.Now()
  182. lastRate := 0.0
  183. didProcessCh := make(chan struct{}, 1)
  184. go func() {
  185. for {
  186. select {
  187. case <-bcR.Quit():
  188. return
  189. case <-bcR.pool.Quit():
  190. return
  191. case request := <-bcR.requestsCh:
  192. peer := bcR.Switch.Peers().Get(request.PeerID)
  193. if peer == nil {
  194. continue
  195. }
  196. msgBytes := cdc.MustMarshalBinaryBare(&bcBlockRequestMessage{request.Height})
  197. queued := peer.TrySend(BlockchainChannel, msgBytes)
  198. if !queued {
  199. bcR.Logger.Debug("Send queue is full, drop block request", "peer", peer.ID(), "height", request.Height)
  200. }
  201. case err := <-bcR.errorsCh:
  202. peer := bcR.Switch.Peers().Get(err.peerID)
  203. if peer != nil {
  204. bcR.Switch.StopPeerForError(peer, err)
  205. }
  206. case <-statusUpdateTicker.C:
  207. // ask for status updates
  208. go bcR.BroadcastStatusRequest() // nolint: errcheck
  209. }
  210. }
  211. }()
  212. FOR_LOOP:
  213. for {
  214. select {
  215. case <-switchToConsensusTicker.C:
  216. height, numPending, lenRequesters := bcR.pool.GetStatus()
  217. outbound, inbound, _ := bcR.Switch.NumPeers()
  218. bcR.Logger.Debug("Consensus ticker", "numPending", numPending, "total", lenRequesters,
  219. "outbound", outbound, "inbound", inbound)
  220. if bcR.pool.IsCaughtUp() {
  221. bcR.Logger.Info("Time to switch to consensus reactor!", "height", height)
  222. bcR.pool.Stop()
  223. conR, ok := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
  224. if ok {
  225. conR.SwitchToConsensus(state, blocksSynced)
  226. }
  227. // else {
  228. // should only happen during testing
  229. // }
  230. break FOR_LOOP
  231. }
  232. case <-trySyncTicker.C: // chan time
  233. select {
  234. case didProcessCh <- struct{}{}:
  235. default:
  236. }
  237. case <-didProcessCh:
  238. // NOTE: It is a subtle mistake to process more than a single block
  239. // at a time (e.g. 10) here, because we only TrySend 1 request per
  240. // loop. The ratio mismatch can result in starving of blocks, a
  241. // sudden burst of requests and responses, and repeat.
  242. // Consequently, it is better to split these routines rather than
  243. // coupling them as it's written here. TODO uncouple from request
  244. // routine.
  245. // See if there are any blocks to sync.
  246. first, second := bcR.pool.PeekTwoBlocks()
  247. //bcR.Logger.Info("TrySync peeked", "first", first, "second", second)
  248. if first == nil || second == nil {
  249. // We need both to sync the first block.
  250. continue FOR_LOOP
  251. } else {
  252. // Try again quickly next loop.
  253. didProcessCh <- struct{}{}
  254. }
  255. firstParts := first.MakePartSet(types.BlockPartSizeBytes)
  256. firstPartsHeader := firstParts.Header()
  257. firstID := types.BlockID{Hash: first.Hash(), PartsHeader: firstPartsHeader}
  258. // Finally, verify the first block using the second's commit
  259. // NOTE: we can probably make this more efficient, but note that calling
  260. // first.Hash() doesn't verify the tx contents, so MakePartSet() is
  261. // currently necessary.
  262. err := state.Validators.VerifyCommit(
  263. chainID, firstID, first.Height, second.LastCommit)
  264. if err != nil {
  265. bcR.Logger.Error("Error in validation", "err", err)
  266. peerID := bcR.pool.RedoRequest(first.Height)
  267. peer := bcR.Switch.Peers().Get(peerID)
  268. if peer != nil {
  269. // NOTE: we've already removed the peer's request, but we
  270. // still need to clean up the rest.
  271. bcR.Switch.StopPeerForError(peer, fmt.Errorf("BlockchainReactor validation error: %v", err))
  272. }
  273. peerID2 := bcR.pool.RedoRequest(second.Height)
  274. peer2 := bcR.Switch.Peers().Get(peerID2)
  275. if peer2 != nil && peer2 != peer {
  276. // NOTE: we've already removed the peer's request, but we
  277. // still need to clean up the rest.
  278. bcR.Switch.StopPeerForError(peer2, fmt.Errorf("BlockchainReactor validation error: %v", err))
  279. }
  280. continue FOR_LOOP
  281. } else {
  282. bcR.pool.PopRequest()
  283. // TODO: batch saves so we dont persist to disk every block
  284. bcR.store.SaveBlock(first, firstParts, second.LastCommit)
  285. // TODO: same thing for app - but we would need a way to
  286. // get the hash without persisting the state
  287. var err error
  288. state, err = bcR.blockExec.ApplyBlock(state, firstID, first)
  289. if err != nil {
  290. // TODO This is bad, are we zombie?
  291. panic(fmt.Sprintf("Failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err))
  292. }
  293. blocksSynced++
  294. if blocksSynced%100 == 0 {
  295. lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
  296. bcR.Logger.Info("Fast Sync Rate", "height", bcR.pool.height,
  297. "max_peer_height", bcR.pool.MaxPeerHeight(), "blocks/s", lastRate)
  298. lastHundred = time.Now()
  299. }
  300. }
  301. continue FOR_LOOP
  302. case <-bcR.Quit():
  303. break FOR_LOOP
  304. }
  305. }
  306. }
  307. // BroadcastStatusRequest broadcasts `BlockStore` height.
  308. func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
  309. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusRequestMessage{bcR.store.Height()})
  310. bcR.Switch.Broadcast(BlockchainChannel, msgBytes)
  311. return nil
  312. }
  313. //-----------------------------------------------------------------------------
  314. // Messages
  315. // BlockchainMessage is a generic message for this reactor.
  316. type BlockchainMessage interface {
  317. ValidateBasic() error
  318. }
  319. // RegisterBlockchainMessages registers the fast sync messages for amino encoding.
  320. func RegisterBlockchainMessages(cdc *amino.Codec) {
  321. cdc.RegisterInterface((*BlockchainMessage)(nil), nil)
  322. cdc.RegisterConcrete(&bcBlockRequestMessage{}, "tendermint/blockchain/BlockRequest", nil)
  323. cdc.RegisterConcrete(&bcBlockResponseMessage{}, "tendermint/blockchain/BlockResponse", nil)
  324. cdc.RegisterConcrete(&bcNoBlockResponseMessage{}, "tendermint/blockchain/NoBlockResponse", nil)
  325. cdc.RegisterConcrete(&bcStatusResponseMessage{}, "tendermint/blockchain/StatusResponse", nil)
  326. cdc.RegisterConcrete(&bcStatusRequestMessage{}, "tendermint/blockchain/StatusRequest", nil)
  327. }
  328. func decodeMsg(bz []byte) (msg BlockchainMessage, err error) {
  329. if len(bz) > maxMsgSize {
  330. return msg, fmt.Errorf("Msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
  331. }
  332. err = cdc.UnmarshalBinaryBare(bz, &msg)
  333. return
  334. }
  335. //-------------------------------------
  336. type bcBlockRequestMessage struct {
  337. Height int64
  338. }
  339. // ValidateBasic performs basic validation.
  340. func (m *bcBlockRequestMessage) ValidateBasic() error {
  341. if m.Height < 0 {
  342. return errors.New("Negative Height")
  343. }
  344. return nil
  345. }
  346. func (m *bcBlockRequestMessage) String() string {
  347. return fmt.Sprintf("[bcBlockRequestMessage %v]", m.Height)
  348. }
  349. type bcNoBlockResponseMessage struct {
  350. Height int64
  351. }
  352. // ValidateBasic performs basic validation.
  353. func (m *bcNoBlockResponseMessage) ValidateBasic() error {
  354. if m.Height < 0 {
  355. return errors.New("Negative Height")
  356. }
  357. return nil
  358. }
  359. func (m *bcNoBlockResponseMessage) String() string {
  360. return fmt.Sprintf("[bcNoBlockResponseMessage %d]", m.Height)
  361. }
  362. //-------------------------------------
  363. type bcBlockResponseMessage struct {
  364. Block *types.Block
  365. }
  366. // ValidateBasic performs basic validation.
  367. func (m *bcBlockResponseMessage) ValidateBasic() error {
  368. return m.Block.ValidateBasic()
  369. }
  370. func (m *bcBlockResponseMessage) String() string {
  371. return fmt.Sprintf("[bcBlockResponseMessage %v]", m.Block.Height)
  372. }
  373. //-------------------------------------
  374. type bcStatusRequestMessage struct {
  375. Height int64
  376. }
  377. // ValidateBasic performs basic validation.
  378. func (m *bcStatusRequestMessage) ValidateBasic() error {
  379. if m.Height < 0 {
  380. return errors.New("Negative Height")
  381. }
  382. return nil
  383. }
  384. func (m *bcStatusRequestMessage) String() string {
  385. return fmt.Sprintf("[bcStatusRequestMessage %v]", m.Height)
  386. }
  387. //-------------------------------------
  388. type bcStatusResponseMessage struct {
  389. Height int64
  390. }
  391. // ValidateBasic performs basic validation.
  392. func (m *bcStatusResponseMessage) ValidateBasic() error {
  393. if m.Height < 0 {
  394. return errors.New("Negative Height")
  395. }
  396. return nil
  397. }
  398. func (m *bcStatusResponseMessage) String() string {
  399. return fmt.Sprintf("[bcStatusResponseMessage %v]", m.Height)
  400. }