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.

400 lines
12 KiB

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
10 years ago
10 years ago
  1. package blockchain
  2. import (
  3. "fmt"
  4. "reflect"
  5. "time"
  6. amino "github.com/tendermint/go-amino"
  7. cmn "github.com/tendermint/tendermint/libs/common"
  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/types"
  12. )
  13. const (
  14. // BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
  15. BlockchainChannel = byte(0x40)
  16. trySyncIntervalMS = 50
  17. // stop syncing when last block's time is
  18. // within this much of the system time.
  19. // stopSyncingDurationMinutes = 10
  20. // ask for best height every 10s
  21. statusUpdateIntervalSeconds = 10
  22. // check if we should switch to consensus reactor
  23. switchToConsensusIntervalSeconds = 1
  24. // NOTE: keep up to date with bcBlockResponseMessage
  25. bcBlockResponseMessagePrefixSize = 4
  26. bcBlockResponseMessageFieldKeySize = 1
  27. maxMsgSize = types.MaxBlockSizeBytes +
  28. bcBlockResponseMessagePrefixSize +
  29. bcBlockResponseMessageFieldKeySize
  30. )
  31. type consensusReactor interface {
  32. // for when we switch from blockchain reactor and fast sync to
  33. // the consensus machine
  34. SwitchToConsensus(sm.State, int)
  35. }
  36. type peerError struct {
  37. err error
  38. peerID p2p.ID
  39. }
  40. func (e peerError) Error() string {
  41. return fmt.Sprintf("error with peer %v: %s", e.peerID, e.err.Error())
  42. }
  43. // BlockchainReactor handles long-term catchup syncing.
  44. type BlockchainReactor struct {
  45. p2p.BaseReactor
  46. // immutable
  47. initialState sm.State
  48. blockExec *sm.BlockExecutor
  49. store *BlockStore
  50. pool *BlockPool
  51. fastSync bool
  52. requestsCh <-chan BlockRequest
  53. errorsCh <-chan peerError
  54. }
  55. // NewBlockchainReactor returns new reactor instance.
  56. func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *BlockStore,
  57. fastSync bool) *BlockchainReactor {
  58. if state.LastBlockHeight != store.Height() {
  59. panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight,
  60. store.Height()))
  61. }
  62. const capacity = 1000 // must be bigger than peers count
  63. requestsCh := make(chan BlockRequest, capacity)
  64. errorsCh := make(chan peerError, capacity) // so we don't block in #Receive#pool.AddBlock
  65. pool := NewBlockPool(
  66. store.Height()+1,
  67. requestsCh,
  68. errorsCh,
  69. )
  70. bcR := &BlockchainReactor{
  71. initialState: state,
  72. blockExec: blockExec,
  73. store: store,
  74. pool: pool,
  75. fastSync: fastSync,
  76. requestsCh: requestsCh,
  77. errorsCh: errorsCh,
  78. }
  79. bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
  80. return bcR
  81. }
  82. // SetLogger implements cmn.Service by setting the logger on reactor and pool.
  83. func (bcR *BlockchainReactor) SetLogger(l log.Logger) {
  84. bcR.BaseService.Logger = l
  85. bcR.pool.Logger = l
  86. }
  87. // OnStart implements cmn.Service.
  88. func (bcR *BlockchainReactor) OnStart() error {
  89. if err := bcR.BaseReactor.OnStart(); err != nil {
  90. return err
  91. }
  92. if bcR.fastSync {
  93. err := bcR.pool.Start()
  94. if err != nil {
  95. return err
  96. }
  97. go bcR.poolRoutine()
  98. }
  99. return nil
  100. }
  101. // OnStop implements cmn.Service.
  102. func (bcR *BlockchainReactor) OnStop() {
  103. bcR.BaseReactor.OnStop()
  104. bcR.pool.Stop()
  105. }
  106. // GetChannels implements Reactor
  107. func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
  108. return []*p2p.ChannelDescriptor{
  109. {
  110. ID: BlockchainChannel,
  111. Priority: 10,
  112. SendQueueCapacity: 1000,
  113. RecvBufferCapacity: 50 * 4096,
  114. RecvMessageCapacity: maxMsgSize,
  115. },
  116. }
  117. }
  118. // AddPeer implements Reactor by sending our state to peer.
  119. func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) {
  120. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusResponseMessage{bcR.store.Height()})
  121. if !peer.Send(BlockchainChannel, msgBytes) {
  122. // doing nothing, will try later in `poolRoutine`
  123. }
  124. // peer is added to the pool once we receive the first
  125. // bcStatusResponseMessage from the peer and call pool.SetPeerHeight
  126. }
  127. // RemovePeer implements Reactor by removing peer from the pool.
  128. func (bcR *BlockchainReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
  129. bcR.pool.RemovePeer(peer.ID())
  130. }
  131. // respondToPeer loads a block and sends it to the requesting peer,
  132. // if we have it. Otherwise, we'll respond saying we don't have it.
  133. // According to the Tendermint spec, if all nodes are honest,
  134. // no node should be requesting for a block that's non-existent.
  135. func (bcR *BlockchainReactor) respondToPeer(msg *bcBlockRequestMessage,
  136. src p2p.Peer) (queued bool) {
  137. block := bcR.store.LoadBlock(msg.Height)
  138. if block != nil {
  139. msgBytes := cdc.MustMarshalBinaryBare(&bcBlockResponseMessage{Block: block})
  140. return src.TrySend(BlockchainChannel, msgBytes)
  141. }
  142. bcR.Logger.Info("Peer asking for a block we don't have", "src", src, "height", msg.Height)
  143. msgBytes := cdc.MustMarshalBinaryBare(&bcNoBlockResponseMessage{Height: msg.Height})
  144. return src.TrySend(BlockchainChannel, msgBytes)
  145. }
  146. // Receive implements Reactor by handling 4 types of messages (look below).
  147. func (bcR *BlockchainReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
  148. msg, err := decodeMsg(msgBytes)
  149. if err != nil {
  150. bcR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
  151. bcR.Switch.StopPeerForError(src, err)
  152. return
  153. }
  154. bcR.Logger.Debug("Receive", "src", src, "chID", chID, "msg", msg)
  155. switch msg := msg.(type) {
  156. case *bcBlockRequestMessage:
  157. if queued := bcR.respondToPeer(msg, src); !queued {
  158. // Unfortunately not queued since the queue is full.
  159. }
  160. case *bcBlockResponseMessage:
  161. // Got a block.
  162. bcR.pool.AddBlock(src.ID(), msg.Block, len(msgBytes))
  163. case *bcStatusRequestMessage:
  164. // Send peer our state.
  165. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusResponseMessage{bcR.store.Height()})
  166. queued := src.TrySend(BlockchainChannel, msgBytes)
  167. if !queued {
  168. // sorry
  169. }
  170. case *bcStatusResponseMessage:
  171. // Got a peer status. Unverified.
  172. bcR.pool.SetPeerHeight(src.ID(), msg.Height)
  173. default:
  174. bcR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
  175. }
  176. }
  177. // Handle messages from the poolReactor telling the reactor what to do.
  178. // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
  179. // (Except for the SYNC_LOOP, which is the primary purpose and must be synchronous.)
  180. func (bcR *BlockchainReactor) poolRoutine() {
  181. trySyncTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
  182. statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
  183. switchToConsensusTicker := time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
  184. blocksSynced := 0
  185. chainID := bcR.initialState.ChainID
  186. state := bcR.initialState
  187. lastHundred := time.Now()
  188. lastRate := 0.0
  189. FOR_LOOP:
  190. for {
  191. select {
  192. case request := <-bcR.requestsCh:
  193. peer := bcR.Switch.Peers().Get(request.PeerID)
  194. if peer == nil {
  195. continue FOR_LOOP // Peer has since been disconnected.
  196. }
  197. msgBytes := cdc.MustMarshalBinaryBare(&bcBlockRequestMessage{request.Height})
  198. queued := peer.TrySend(BlockchainChannel, msgBytes)
  199. if !queued {
  200. // We couldn't make the request, send-queue full.
  201. // The pool handles timeouts, just let it go.
  202. continue FOR_LOOP
  203. }
  204. case err := <-bcR.errorsCh:
  205. peer := bcR.Switch.Peers().Get(err.peerID)
  206. if peer != nil {
  207. bcR.Switch.StopPeerForError(peer, err)
  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. peerID := bcR.pool.RedoRequest(first.Height)
  247. peer := bcR.Switch.Peers().Get(peerID)
  248. if peer != nil {
  249. bcR.Switch.StopPeerForError(peer, fmt.Errorf("BlockchainReactor validation error: %v", err))
  250. }
  251. break SYNC_LOOP
  252. } else {
  253. bcR.pool.PopRequest()
  254. // TODO: batch saves so we dont persist to disk every block
  255. bcR.store.SaveBlock(first, firstParts, second.LastCommit)
  256. // TODO: same thing for app - but we would need a way to
  257. // get the hash without persisting the state
  258. var err error
  259. state, err = bcR.blockExec.ApplyBlock(state, firstID, first)
  260. if err != nil {
  261. // TODO This is bad, are we zombie?
  262. cmn.PanicQ(cmn.Fmt("Failed to process committed block (%d:%X): %v",
  263. first.Height, first.Hash(), err))
  264. }
  265. blocksSynced++
  266. if blocksSynced%100 == 0 {
  267. lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
  268. bcR.Logger.Info("Fast Sync Rate", "height", bcR.pool.height,
  269. "max_peer_height", bcR.pool.MaxPeerHeight(), "blocks/s", lastRate)
  270. lastHundred = time.Now()
  271. }
  272. }
  273. }
  274. continue FOR_LOOP
  275. case <-bcR.Quit():
  276. break FOR_LOOP
  277. }
  278. }
  279. }
  280. // BroadcastStatusRequest broadcasts `BlockStore` height.
  281. func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
  282. msgBytes := cdc.MustMarshalBinaryBare(&bcStatusRequestMessage{bcR.store.Height()})
  283. bcR.Switch.Broadcast(BlockchainChannel, msgBytes)
  284. return nil
  285. }
  286. //-----------------------------------------------------------------------------
  287. // Messages
  288. // BlockchainMessage is a generic message for this reactor.
  289. type BlockchainMessage interface{}
  290. func RegisterBlockchainMessages(cdc *amino.Codec) {
  291. cdc.RegisterInterface((*BlockchainMessage)(nil), nil)
  292. cdc.RegisterConcrete(&bcBlockRequestMessage{}, "tendermint/mempool/BlockRequest", nil)
  293. cdc.RegisterConcrete(&bcBlockResponseMessage{}, "tendermint/mempool/BlockResponse", nil)
  294. cdc.RegisterConcrete(&bcNoBlockResponseMessage{}, "tendermint/mempool/NoBlockResponse", nil)
  295. cdc.RegisterConcrete(&bcStatusResponseMessage{}, "tendermint/mempool/StatusResponse", nil)
  296. cdc.RegisterConcrete(&bcStatusRequestMessage{}, "tendermint/mempool/StatusRequest", nil)
  297. }
  298. func decodeMsg(bz []byte) (msg BlockchainMessage, err error) {
  299. if len(bz) > maxMsgSize {
  300. return msg, fmt.Errorf("Msg exceeds max size (%d > %d)", len(bz), maxMsgSize)
  301. }
  302. err = cdc.UnmarshalBinaryBare(bz, &msg)
  303. return
  304. }
  305. //-------------------------------------
  306. type bcBlockRequestMessage struct {
  307. Height int64
  308. }
  309. func (m *bcBlockRequestMessage) String() string {
  310. return cmn.Fmt("[bcBlockRequestMessage %v]", m.Height)
  311. }
  312. type bcNoBlockResponseMessage struct {
  313. Height int64
  314. }
  315. func (brm *bcNoBlockResponseMessage) String() string {
  316. return cmn.Fmt("[bcNoBlockResponseMessage %d]", brm.Height)
  317. }
  318. //-------------------------------------
  319. type bcBlockResponseMessage struct {
  320. Block *types.Block
  321. }
  322. func (m *bcBlockResponseMessage) String() string {
  323. return cmn.Fmt("[bcBlockResponseMessage %v]", m.Block.Height)
  324. }
  325. //-------------------------------------
  326. type bcStatusRequestMessage struct {
  327. Height int64
  328. }
  329. func (m *bcStatusRequestMessage) String() string {
  330. return cmn.Fmt("[bcStatusRequestMessage %v]", m.Height)
  331. }
  332. //-------------------------------------
  333. type bcStatusResponseMessage struct {
  334. Height int64
  335. }
  336. func (m *bcStatusResponseMessage) String() string {
  337. return cmn.Fmt("[bcStatusResponseMessage %v]", m.Height)
  338. }