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.

405 lines
12 KiB

8 years ago
8 years ago
7 years ago
7 years ago
7 years ago
8 years ago
10 years ago
8 years ago
8 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
  1. package blockchain
  2. import (
  3. "bytes"
  4. "errors"
  5. "reflect"
  6. "sync"
  7. "time"
  8. wire "github.com/tendermint/go-wire"
  9. "github.com/tendermint/tendermint/p2p"
  10. sm "github.com/tendermint/tendermint/state"
  11. "github.com/tendermint/tendermint/types"
  12. cmn "github.com/tendermint/tmlibs/common"
  13. "github.com/tendermint/tmlibs/log"
  14. )
  15. const (
  16. // BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
  17. BlockchainChannel = byte(0x40)
  18. defaultChannelCapacity = 1000
  19. trySyncIntervalMS = 50
  20. // stop syncing when last block's time is
  21. // within this much of the system time.
  22. // stopSyncingDurationMinutes = 10
  23. // ask for best height every 10s
  24. statusUpdateIntervalSeconds = 10
  25. // check if we should switch to consensus reactor
  26. switchToConsensusIntervalSeconds = 1
  27. )
  28. type consensusReactor interface {
  29. // for when we switch from blockchain reactor and fast sync to
  30. // the consensus machine
  31. SwitchToConsensus(sm.State, int)
  32. }
  33. // BlockchainReactor handles long-term catchup syncing.
  34. type BlockchainReactor struct {
  35. p2p.BaseReactor
  36. mtx sync.Mutex
  37. params types.ConsensusParams
  38. // immutable
  39. initialState sm.State
  40. blockExec *sm.BlockExecutor
  41. store *BlockStore
  42. pool *BlockPool
  43. fastSync bool
  44. requestsCh chan BlockRequest
  45. timeoutsCh chan p2p.ID
  46. }
  47. // NewBlockchainReactor returns new reactor instance.
  48. func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *BlockStore, fastSync bool) *BlockchainReactor {
  49. if state.LastBlockHeight != store.Height() {
  50. cmn.PanicSanity(cmn.Fmt("state (%v) and store (%v) height mismatch", state.LastBlockHeight, store.Height()))
  51. }
  52. requestsCh := make(chan BlockRequest, defaultChannelCapacity)
  53. timeoutsCh := make(chan p2p.ID, defaultChannelCapacity)
  54. pool := NewBlockPool(
  55. store.Height()+1,
  56. requestsCh,
  57. timeoutsCh,
  58. )
  59. bcR := &BlockchainReactor{
  60. params: state.ConsensusParams,
  61. initialState: state,
  62. blockExec: blockExec,
  63. store: store,
  64. pool: pool,
  65. fastSync: fastSync,
  66. requestsCh: requestsCh,
  67. timeoutsCh: timeoutsCh,
  68. }
  69. bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
  70. return bcR
  71. }
  72. // SetLogger implements cmn.Service by setting the logger on reactor and pool.
  73. func (bcR *BlockchainReactor) SetLogger(l log.Logger) {
  74. bcR.BaseService.Logger = l
  75. bcR.pool.Logger = l
  76. }
  77. // OnStart implements cmn.Service.
  78. func (bcR *BlockchainReactor) OnStart() error {
  79. if err := bcR.BaseReactor.OnStart(); err != nil {
  80. return err
  81. }
  82. if bcR.fastSync {
  83. err := bcR.pool.Start()
  84. if err != nil {
  85. return err
  86. }
  87. go bcR.poolRoutine()
  88. }
  89. return nil
  90. }
  91. // OnStop implements cmn.Service.
  92. func (bcR *BlockchainReactor) OnStop() {
  93. bcR.BaseReactor.OnStop()
  94. bcR.pool.Stop()
  95. }
  96. // GetChannels implements Reactor
  97. func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
  98. return []*p2p.ChannelDescriptor{
  99. {
  100. ID: BlockchainChannel,
  101. Priority: 10,
  102. SendQueueCapacity: 1000,
  103. },
  104. }
  105. }
  106. // AddPeer implements Reactor by sending our state to peer.
  107. func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) {
  108. if !peer.Send(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusResponseMessage{bcR.store.Height()}}) {
  109. // doing nothing, will try later in `poolRoutine`
  110. }
  111. // peer is added to the pool once we receive the first
  112. // bcStatusResponseMessage from the peer and call pool.SetPeerHeight
  113. }
  114. // RemovePeer implements Reactor by removing peer from the pool.
  115. func (bcR *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
  116. bcR.pool.RemovePeer(peer.ID())
  117. }
  118. // respondToPeer loads a block and sends it to the requesting peer,
  119. // if we have it. Otherwise, we'll respond saying we don't have it.
  120. // According to the Tendermint spec, if all nodes are honest,
  121. // no node should be requesting for a block that's non-existent.
  122. func (bcR *BlockchainReactor) respondToPeer(msg *bcBlockRequestMessage, src p2p.Peer) (queued bool) {
  123. block := bcR.store.LoadBlock(msg.Height)
  124. if block != nil {
  125. msg := &bcBlockResponseMessage{Block: block}
  126. return src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{msg})
  127. }
  128. bcR.Logger.Info("Peer asking for a block we don't have", "src", src, "height", msg.Height)
  129. return src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{
  130. &bcNoBlockResponseMessage{Height: msg.Height},
  131. })
  132. }
  133. // Receive implements Reactor by handling 4 types of messages (look below).
  134. func (bcR *BlockchainReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  135. _, msg, err := DecodeMessage(msgBytes, bcR.maxMsgSize())
  136. if err != nil {
  137. bcR.Logger.Error("Error decoding message", "err", err)
  138. return
  139. }
  140. bcR.Logger.Debug("Receive", "src", src, "chID", chID, "msg", msg)
  141. // TODO: improve logic to satisfy megacheck
  142. switch msg := msg.(type) {
  143. case *bcBlockRequestMessage:
  144. if queued := bcR.respondToPeer(msg, src); !queued {
  145. // Unfortunately not queued since the queue is full.
  146. }
  147. case *bcBlockResponseMessage:
  148. // Got a block.
  149. bcR.pool.AddBlock(src.ID(), msg.Block, len(msgBytes))
  150. case *bcStatusRequestMessage:
  151. // Send peer our state.
  152. queued := src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusResponseMessage{bcR.store.Height()}})
  153. if !queued {
  154. // sorry
  155. }
  156. case *bcStatusResponseMessage:
  157. // Got a peer status. Unverified.
  158. bcR.pool.SetPeerHeight(src.ID(), msg.Height)
  159. default:
  160. bcR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
  161. }
  162. }
  163. // maxMsgSize returns the maximum allowable size of a
  164. // message on the blockchain reactor.
  165. func (bcR *BlockchainReactor) maxMsgSize() int {
  166. bcR.mtx.Lock()
  167. defer bcR.mtx.Unlock()
  168. return bcR.params.BlockSize.MaxBytes + 2
  169. }
  170. // updateConsensusParams updates the internal consensus params
  171. func (bcR *BlockchainReactor) updateConsensusParams(params types.ConsensusParams) {
  172. bcR.mtx.Lock()
  173. defer bcR.mtx.Unlock()
  174. bcR.params = params
  175. }
  176. // Handle messages from the poolReactor telling the reactor what to do.
  177. // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
  178. // (Except for the SYNC_LOOP, which is the primary purpose and must be synchronous.)
  179. func (bcR *BlockchainReactor) poolRoutine() {
  180. trySyncTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
  181. statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
  182. switchToConsensusTicker := time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
  183. blocksSynced := 0
  184. chainID := bcR.initialState.ChainID
  185. state := bcR.initialState
  186. lastHundred := time.Now()
  187. lastRate := 0.0
  188. FOR_LOOP:
  189. for {
  190. select {
  191. case request := <-bcR.requestsCh: // chan BlockRequest
  192. peer := bcR.Switch.Peers().Get(request.PeerID)
  193. if peer == nil {
  194. continue FOR_LOOP // Peer has since been disconnected.
  195. }
  196. msg := &bcBlockRequestMessage{request.Height}
  197. queued := peer.TrySend(BlockchainChannel, struct{ BlockchainMessage }{msg})
  198. if !queued {
  199. // We couldn't make the request, send-queue full.
  200. // The pool handles timeouts, just let it go.
  201. continue FOR_LOOP
  202. }
  203. case peerID := <-bcR.timeoutsCh: // chan string
  204. // Peer timed out.
  205. peer := bcR.Switch.Peers().Get(peerID)
  206. if peer != nil {
  207. bcR.Switch.StopPeerForError(peer, errors.New("BlockchainReactor Timeout"))
  208. }
  209. case <-statusUpdateTicker.C:
  210. // ask for status updates
  211. go bcR.BroadcastStatusRequest() // nolint: errcheck
  212. case <-switchToConsensusTicker.C:
  213. height, numPending, lenRequesters := bcR.pool.GetStatus()
  214. outbound, inbound, _ := bcR.Switch.NumPeers()
  215. bcR.Logger.Debug("Consensus ticker", "numPending", numPending, "total", lenRequesters,
  216. "outbound", outbound, "inbound", inbound)
  217. if bcR.pool.IsCaughtUp() {
  218. bcR.Logger.Info("Time to switch to consensus reactor!", "height", height)
  219. bcR.pool.Stop()
  220. conR := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
  221. conR.SwitchToConsensus(state, blocksSynced)
  222. break FOR_LOOP
  223. }
  224. case <-trySyncTicker.C: // chan time
  225. // This loop can be slow as long as it's doing syncing work.
  226. SYNC_LOOP:
  227. for i := 0; i < 10; i++ {
  228. // See if there are any blocks to sync.
  229. first, second := bcR.pool.PeekTwoBlocks()
  230. //bcR.Logger.Info("TrySync peeked", "first", first, "second", second)
  231. if first == nil || second == nil {
  232. // We need both to sync the first block.
  233. break SYNC_LOOP
  234. }
  235. firstParts := first.MakePartSet(state.ConsensusParams.BlockPartSizeBytes)
  236. firstPartsHeader := firstParts.Header()
  237. firstID := types.BlockID{first.Hash(), firstPartsHeader}
  238. // Finally, verify the first block using the second's commit
  239. // NOTE: we can probably make this more efficient, but note that calling
  240. // first.Hash() doesn't verify the tx contents, so MakePartSet() is
  241. // currently necessary.
  242. err := state.Validators.VerifyCommit(
  243. chainID, firstID, first.Height, second.LastCommit)
  244. if err != nil {
  245. bcR.Logger.Error("Error in validation", "err", err)
  246. bcR.pool.RedoRequest(first.Height)
  247. break SYNC_LOOP
  248. } else {
  249. bcR.pool.PopRequest()
  250. bcR.store.SaveBlock(first, firstParts, second.LastCommit)
  251. // NOTE: we could improve performance if we
  252. // didn't make the app commit to disk every block
  253. // ... but we would need a way to get the hash without it persisting
  254. var err error
  255. state, err = bcR.blockExec.ApplyBlock(state, firstID, first)
  256. if err != nil {
  257. // TODO This is bad, are we zombie?
  258. cmn.PanicQ(cmn.Fmt("Failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err))
  259. }
  260. blocksSynced += 1
  261. // update the consensus params
  262. bcR.updateConsensusParams(state.ConsensusParams)
  263. if blocksSynced%100 == 0 {
  264. lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
  265. bcR.Logger.Info("Fast Sync Rate", "height", bcR.pool.height,
  266. "max_peer_height", bcR.pool.MaxPeerHeight(), "blocks/s", lastRate)
  267. lastHundred = time.Now()
  268. }
  269. }
  270. }
  271. continue FOR_LOOP
  272. case <-bcR.Quit:
  273. break FOR_LOOP
  274. }
  275. }
  276. }
  277. // BroadcastStatusRequest broadcasts `BlockStore` height.
  278. func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
  279. bcR.Switch.Broadcast(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusRequestMessage{bcR.store.Height()}})
  280. return nil
  281. }
  282. //-----------------------------------------------------------------------------
  283. // Messages
  284. const (
  285. msgTypeBlockRequest = byte(0x10)
  286. msgTypeBlockResponse = byte(0x11)
  287. msgTypeNoBlockResponse = byte(0x12)
  288. msgTypeStatusResponse = byte(0x20)
  289. msgTypeStatusRequest = byte(0x21)
  290. )
  291. // BlockchainMessage is a generic message for this reactor.
  292. type BlockchainMessage interface{}
  293. var _ = wire.RegisterInterface(
  294. struct{ BlockchainMessage }{},
  295. wire.ConcreteType{&bcBlockRequestMessage{}, msgTypeBlockRequest},
  296. wire.ConcreteType{&bcBlockResponseMessage{}, msgTypeBlockResponse},
  297. wire.ConcreteType{&bcNoBlockResponseMessage{}, msgTypeNoBlockResponse},
  298. wire.ConcreteType{&bcStatusResponseMessage{}, msgTypeStatusResponse},
  299. wire.ConcreteType{&bcStatusRequestMessage{}, msgTypeStatusRequest},
  300. )
  301. // DecodeMessage decodes BlockchainMessage.
  302. // TODO: ensure that bz is completely read.
  303. func DecodeMessage(bz []byte, maxSize int) (msgType byte, msg BlockchainMessage, err error) {
  304. msgType = bz[0]
  305. n := int(0)
  306. r := bytes.NewReader(bz)
  307. msg = wire.ReadBinary(struct{ BlockchainMessage }{}, r, maxSize, &n, &err).(struct{ BlockchainMessage }).BlockchainMessage
  308. if err != nil && n != len(bz) {
  309. err = errors.New("DecodeMessage() had bytes left over")
  310. }
  311. return
  312. }
  313. //-------------------------------------
  314. type bcBlockRequestMessage struct {
  315. Height int64
  316. }
  317. func (m *bcBlockRequestMessage) String() string {
  318. return cmn.Fmt("[bcBlockRequestMessage %v]", m.Height)
  319. }
  320. type bcNoBlockResponseMessage struct {
  321. Height int64
  322. }
  323. func (brm *bcNoBlockResponseMessage) String() string {
  324. return cmn.Fmt("[bcNoBlockResponseMessage %d]", brm.Height)
  325. }
  326. //-------------------------------------
  327. // NOTE: keep up-to-date with maxBlockchainResponseSize
  328. type bcBlockResponseMessage struct {
  329. Block *types.Block
  330. }
  331. func (m *bcBlockResponseMessage) String() string {
  332. return cmn.Fmt("[bcBlockResponseMessage %v]", m.Block.Height)
  333. }
  334. //-------------------------------------
  335. type bcStatusRequestMessage struct {
  336. Height int64
  337. }
  338. func (m *bcStatusRequestMessage) String() string {
  339. return cmn.Fmt("[bcStatusRequestMessage %v]", m.Height)
  340. }
  341. //-------------------------------------
  342. type bcStatusResponseMessage struct {
  343. Height int64
  344. }
  345. func (m *bcStatusResponseMessage) String() string {
  346. return cmn.Fmt("[bcStatusResponseMessage %v]", m.Height)
  347. }