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.

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