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.

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