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.

348 lines
11 KiB

10 years ago
10 years ago
8 years ago
8 years ago
10 years ago
10 years ago
10 years ago
10 years ago
  1. package blockchain
  2. import (
  3. "bytes"
  4. "errors"
  5. "reflect"
  6. "time"
  7. "github.com/spf13/viper"
  8. "github.com/tendermint/go-wire"
  9. "github.com/tendermint/tendermint/p2p"
  10. "github.com/tendermint/tendermint/proxy"
  11. sm "github.com/tendermint/tendermint/state"
  12. "github.com/tendermint/tendermint/types"
  13. cmn "github.com/tendermint/tmlibs/common"
  14. )
  15. const (
  16. // BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
  17. BlockchainChannel = byte(0x40)
  18. defaultChannelCapacity = 100
  19. defaultSleepIntervalMS = 500
  20. trySyncIntervalMS = 100
  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. maxBlockchainResponseSize = types.MaxBlockSize + 2
  29. )
  30. type consensusReactor interface {
  31. // for when we switch from blockchain reactor and fast sync to
  32. // the consensus machine
  33. SwitchToConsensus(*sm.State)
  34. }
  35. // BlockchainReactor handles long-term catchup syncing.
  36. type BlockchainReactor struct {
  37. p2p.BaseReactor
  38. config *viper.Viper
  39. state *sm.State
  40. proxyAppConn proxy.AppConnConsensus // same as consensus.proxyAppConn
  41. store *BlockStore
  42. pool *BlockPool
  43. fastSync bool
  44. requestsCh chan BlockRequest
  45. timeoutsCh chan string
  46. lastBlock *types.Block
  47. evsw types.EventSwitch
  48. }
  49. // NewBlockchainReactor returns new reactor instance.
  50. func NewBlockchainReactor(config *viper.Viper, state *sm.State, proxyAppConn proxy.AppConnConsensus, store *BlockStore, fastSync bool) *BlockchainReactor {
  51. if state.LastBlockHeight == store.Height()-1 {
  52. store.height-- // XXX HACK, make this better
  53. }
  54. if state.LastBlockHeight != store.Height() {
  55. cmn.PanicSanity(cmn.Fmt("state (%v) and store (%v) height mismatch", state.LastBlockHeight, store.Height()))
  56. }
  57. requestsCh := make(chan BlockRequest, defaultChannelCapacity)
  58. timeoutsCh := make(chan string, defaultChannelCapacity)
  59. pool := NewBlockPool(
  60. store.Height()+1,
  61. requestsCh,
  62. timeoutsCh,
  63. )
  64. bcR := &BlockchainReactor{
  65. config: config,
  66. state: state,
  67. proxyAppConn: proxyAppConn,
  68. store: store,
  69. pool: pool,
  70. fastSync: fastSync,
  71. requestsCh: requestsCh,
  72. timeoutsCh: timeoutsCh,
  73. }
  74. bcR.BaseReactor = *p2p.NewBaseReactor(log, "BlockchainReactor", bcR)
  75. return bcR
  76. }
  77. // OnStart implements BaseService
  78. func (bcR *BlockchainReactor) OnStart() error {
  79. bcR.BaseReactor.OnStart()
  80. if bcR.fastSync {
  81. _, err := bcR.pool.Start()
  82. if err != nil {
  83. return err
  84. }
  85. go bcR.poolRoutine()
  86. }
  87. return nil
  88. }
  89. // OnStop implements BaseService
  90. func (bcR *BlockchainReactor) OnStop() {
  91. bcR.BaseReactor.OnStop()
  92. bcR.pool.Stop()
  93. }
  94. // GetChannels implements Reactor
  95. func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
  96. return []*p2p.ChannelDescriptor{
  97. &p2p.ChannelDescriptor{
  98. ID: BlockchainChannel,
  99. Priority: 5,
  100. SendQueueCapacity: 100,
  101. },
  102. }
  103. }
  104. // AddPeer implements Reactor by sending our state to peer.
  105. func (bcR *BlockchainReactor) AddPeer(peer *p2p.Peer) {
  106. if !peer.Send(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusResponseMessage{bcR.store.Height()}}) {
  107. // doing nothing, will try later in `poolRoutine`
  108. }
  109. }
  110. // RemovePeer implements Reactor by removing peer from the pool.
  111. func (bcR *BlockchainReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
  112. bcR.pool.RemovePeer(peer.Key)
  113. }
  114. // Receive implements Reactor by handling 4 types of messages (look below).
  115. func (bcR *BlockchainReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
  116. _, msg, err := DecodeMessage(msgBytes)
  117. if err != nil {
  118. log.Warn("Error decoding message", "error", err)
  119. return
  120. }
  121. log.Debug("Receive", "src", src, "chID", chID, "msg", msg)
  122. switch msg := msg.(type) {
  123. case *bcBlockRequestMessage:
  124. // Got a request for a block. Respond with block if we have it.
  125. block := bcR.store.LoadBlock(msg.Height)
  126. if block != nil {
  127. msg := &bcBlockResponseMessage{Block: block}
  128. queued := src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{msg})
  129. if !queued {
  130. // queue is full, just ignore.
  131. }
  132. } else {
  133. // TODO peer is asking for things we don't have.
  134. }
  135. case *bcBlockResponseMessage:
  136. // Got a block.
  137. bcR.pool.AddBlock(src.Key, msg.Block, len(msgBytes))
  138. case *bcStatusRequestMessage:
  139. // Send peer our state.
  140. queued := src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusResponseMessage{bcR.store.Height()}})
  141. if !queued {
  142. // sorry
  143. }
  144. case *bcStatusResponseMessage:
  145. // Got a peer status. Unverified.
  146. bcR.pool.SetPeerHeight(src.Key, msg.Height)
  147. default:
  148. log.Warn(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
  149. }
  150. }
  151. // Handle messages from the poolReactor telling the reactor what to do.
  152. // NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
  153. // (Except for the SYNC_LOOP, which is the primary purpose and must be synchronous.)
  154. func (bcR *BlockchainReactor) poolRoutine() {
  155. trySyncTicker := time.NewTicker(trySyncIntervalMS * time.Millisecond)
  156. statusUpdateTicker := time.NewTicker(statusUpdateIntervalSeconds * time.Second)
  157. switchToConsensusTicker := time.NewTicker(switchToConsensusIntervalSeconds * time.Second)
  158. FOR_LOOP:
  159. for {
  160. select {
  161. case request := <-bcR.requestsCh: // chan BlockRequest
  162. peer := bcR.Switch.Peers().Get(request.PeerID)
  163. if peer == nil {
  164. continue FOR_LOOP // Peer has since been disconnected.
  165. }
  166. msg := &bcBlockRequestMessage{request.Height}
  167. queued := peer.TrySend(BlockchainChannel, struct{ BlockchainMessage }{msg})
  168. if !queued {
  169. // We couldn't make the request, send-queue full.
  170. // The pool handles timeouts, just let it go.
  171. continue FOR_LOOP
  172. }
  173. case peerID := <-bcR.timeoutsCh: // chan string
  174. // Peer timed out.
  175. peer := bcR.Switch.Peers().Get(peerID)
  176. if peer != nil {
  177. bcR.Switch.StopPeerForError(peer, errors.New("BlockchainReactor Timeout"))
  178. }
  179. case _ = <-statusUpdateTicker.C:
  180. // ask for status updates
  181. go bcR.BroadcastStatusRequest()
  182. case _ = <-switchToConsensusTicker.C:
  183. height, numPending, _ := bcR.pool.GetStatus()
  184. outbound, inbound, _ := bcR.Switch.NumPeers()
  185. log.Info("Consensus ticker", "numPending", numPending, "total", len(bcR.pool.requesters),
  186. "outbound", outbound, "inbound", inbound)
  187. if bcR.pool.IsCaughtUp() {
  188. log.Notice("Time to switch to consensus reactor!", "height", height)
  189. bcR.pool.Stop()
  190. conR := bcR.Switch.Reactor("CONSENSUS").(consensusReactor)
  191. conR.SwitchToConsensus(bcR.state)
  192. break FOR_LOOP
  193. }
  194. case _ = <-trySyncTicker.C: // chan time
  195. // This loop can be slow as long as it's doing syncing work.
  196. SYNC_LOOP:
  197. for i := 0; i < 10; i++ {
  198. // See if there are any blocks to sync.
  199. first, second := bcR.pool.PeekTwoBlocks()
  200. //log.Info("TrySync peeked", "first", first, "second", second)
  201. if first == nil || second == nil {
  202. // We need both to sync the first block.
  203. break SYNC_LOOP
  204. }
  205. firstParts := first.MakePartSet(bcR.config.GetInt("block_part_size")) // TODO: put part size in parts header?
  206. firstPartsHeader := firstParts.Header()
  207. // Finally, verify the first block using the second's commit
  208. // NOTE: we can probably make this more efficient, but note that calling
  209. // first.Hash() doesn't verify the tx contents, so MakePartSet() is
  210. // currently necessary.
  211. err := bcR.state.Validators.VerifyCommit(
  212. bcR.state.ChainID, types.BlockID{first.Hash(), firstPartsHeader}, first.Height, second.LastCommit)
  213. if err != nil {
  214. log.Info("error in validation", "error", err)
  215. bcR.pool.RedoRequest(first.Height)
  216. break SYNC_LOOP
  217. } else {
  218. bcR.pool.PopRequest()
  219. bcR.store.SaveBlock(first, firstParts, second.LastCommit)
  220. // TODO: should we be firing events? need to fire NewBlock events manually ...
  221. // NOTE: we could improve performance if we
  222. // didn't make the app commit to disk every block
  223. // ... but we would need a way to get the hash without it persisting
  224. err := bcR.state.ApplyBlock(bcR.evsw, bcR.proxyAppConn, first, firstPartsHeader, types.MockMempool{})
  225. if err != nil {
  226. // TODO This is bad, are we zombie?
  227. cmn.PanicQ(cmn.Fmt("Failed to process committed block (%d:%X): %v", first.Height, first.Hash(), err))
  228. }
  229. }
  230. }
  231. continue FOR_LOOP
  232. case <-bcR.Quit:
  233. break FOR_LOOP
  234. }
  235. }
  236. }
  237. // BroadcastStatusRequest broadcasts `BlockStore` height.
  238. func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
  239. bcR.Switch.Broadcast(BlockchainChannel, struct{ BlockchainMessage }{&bcStatusRequestMessage{bcR.store.Height()}})
  240. return nil
  241. }
  242. // SetEventSwitch implements events.Eventable
  243. func (bcR *BlockchainReactor) SetEventSwitch(evsw types.EventSwitch) {
  244. bcR.evsw = evsw
  245. }
  246. //-----------------------------------------------------------------------------
  247. // Messages
  248. const (
  249. msgTypeBlockRequest = byte(0x10)
  250. msgTypeBlockResponse = byte(0x11)
  251. msgTypeStatusResponse = byte(0x20)
  252. msgTypeStatusRequest = byte(0x21)
  253. )
  254. // BlockchainMessage is a generic message for this reactor.
  255. type BlockchainMessage interface{}
  256. var _ = wire.RegisterInterface(
  257. struct{ BlockchainMessage }{},
  258. wire.ConcreteType{&bcBlockRequestMessage{}, msgTypeBlockRequest},
  259. wire.ConcreteType{&bcBlockResponseMessage{}, msgTypeBlockResponse},
  260. wire.ConcreteType{&bcStatusResponseMessage{}, msgTypeStatusResponse},
  261. wire.ConcreteType{&bcStatusRequestMessage{}, msgTypeStatusRequest},
  262. )
  263. // DecodeMessage decodes BlockchainMessage.
  264. // TODO: ensure that bz is completely read.
  265. func DecodeMessage(bz []byte) (msgType byte, msg BlockchainMessage, err error) {
  266. msgType = bz[0]
  267. n := int(0)
  268. r := bytes.NewReader(bz)
  269. msg = wire.ReadBinary(struct{ BlockchainMessage }{}, r, maxBlockchainResponseSize, &n, &err).(struct{ BlockchainMessage }).BlockchainMessage
  270. if err != nil && n != len(bz) {
  271. err = errors.New("DecodeMessage() had bytes left over")
  272. }
  273. return
  274. }
  275. //-------------------------------------
  276. type bcBlockRequestMessage struct {
  277. Height int
  278. }
  279. func (m *bcBlockRequestMessage) String() string {
  280. return cmn.Fmt("[bcBlockRequestMessage %v]", m.Height)
  281. }
  282. //-------------------------------------
  283. // NOTE: keep up-to-date with maxBlockchainResponseSize
  284. type bcBlockResponseMessage struct {
  285. Block *types.Block
  286. }
  287. func (m *bcBlockResponseMessage) String() string {
  288. return cmn.Fmt("[bcBlockResponseMessage %v]", m.Block.Height)
  289. }
  290. //-------------------------------------
  291. type bcStatusRequestMessage struct {
  292. Height int
  293. }
  294. func (m *bcStatusRequestMessage) String() string {
  295. return cmn.Fmt("[bcStatusRequestMessage %v]", m.Height)
  296. }
  297. //-------------------------------------
  298. type bcStatusResponseMessage struct {
  299. Height int
  300. }
  301. func (m *bcStatusResponseMessage) String() string {
  302. return cmn.Fmt("[bcStatusResponseMessage %v]", m.Height)
  303. }