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.

201 lines
6.4 KiB

7 years ago
7 years ago
  1. package consensus
  2. import (
  3. "bufio"
  4. "bytes"
  5. "fmt"
  6. "math/rand"
  7. "os"
  8. "path/filepath"
  9. "strings"
  10. "time"
  11. "github.com/pkg/errors"
  12. "github.com/tendermint/abci/example/kvstore"
  13. bc "github.com/tendermint/tendermint/blockchain"
  14. cfg "github.com/tendermint/tendermint/config"
  15. "github.com/tendermint/tendermint/proxy"
  16. sm "github.com/tendermint/tendermint/state"
  17. "github.com/tendermint/tendermint/types"
  18. auto "github.com/tendermint/tmlibs/autofile"
  19. cmn "github.com/tendermint/tmlibs/common"
  20. "github.com/tendermint/tmlibs/db"
  21. "github.com/tendermint/tmlibs/log"
  22. )
  23. // WALWithNBlocks generates a consensus WAL. It does this by spining up a
  24. // stripped down version of node (proxy app, event bus, consensus state) with a
  25. // persistent kvstore application and special consensus wal instance
  26. // (byteBufferWAL) and waits until numBlocks are created. Then it returns a WAL
  27. // content.
  28. func WALWithNBlocks(numBlocks int) (data []byte, err error) {
  29. config := getConfig()
  30. app := kvstore.NewPersistentKVStoreApplication(filepath.Join(config.DBDir(), "wal_generator"))
  31. logger := log.TestingLogger().With("wal_generator", "wal_generator")
  32. logger.Info("generating WAL (last height msg excluded)", "numBlocks", numBlocks)
  33. /////////////////////////////////////////////////////////////////////////////
  34. // COPY PASTE FROM node.go WITH A FEW MODIFICATIONS
  35. // NOTE: we can't import node package because of circular dependency
  36. privValidatorFile := config.PrivValidatorFile()
  37. privValidator := types.LoadOrGenPrivValidatorFS(privValidatorFile)
  38. genDoc, err := types.GenesisDocFromFile(config.GenesisFile())
  39. if err != nil {
  40. return nil, errors.Wrap(err, "failed to read genesis file")
  41. }
  42. stateDB := db.NewMemDB()
  43. blockStoreDB := db.NewMemDB()
  44. state, err := sm.MakeGenesisState(genDoc)
  45. if err != nil {
  46. return nil, errors.Wrap(err, "failed to make genesis state")
  47. }
  48. blockStore := bc.NewBlockStore(blockStoreDB)
  49. handshaker := NewHandshaker(stateDB, state, blockStore, genDoc.AppState())
  50. proxyApp := proxy.NewAppConns(proxy.NewLocalClientCreator(app), handshaker)
  51. proxyApp.SetLogger(logger.With("module", "proxy"))
  52. if err := proxyApp.Start(); err != nil {
  53. return nil, errors.Wrap(err, "failed to start proxy app connections")
  54. }
  55. defer proxyApp.Stop()
  56. eventBus := types.NewEventBus()
  57. eventBus.SetLogger(logger.With("module", "events"))
  58. if err := eventBus.Start(); err != nil {
  59. return nil, errors.Wrap(err, "failed to start event bus")
  60. }
  61. defer eventBus.Stop()
  62. mempool := types.MockMempool{}
  63. evpool := types.MockEvidencePool{}
  64. blockExec := sm.NewBlockExecutor(stateDB, log.TestingLogger(), proxyApp.Consensus(), mempool, evpool)
  65. consensusState := NewConsensusState(config.Consensus, state.Copy(), blockExec, blockStore, mempool, evpool)
  66. consensusState.SetLogger(logger)
  67. consensusState.SetEventBus(eventBus)
  68. if privValidator != nil {
  69. consensusState.SetPrivValidator(privValidator)
  70. }
  71. // END OF COPY PASTE
  72. /////////////////////////////////////////////////////////////////////////////
  73. // set consensus wal to buffered WAL, which will write all incoming msgs to buffer
  74. var b bytes.Buffer
  75. wr := bufio.NewWriter(&b)
  76. numBlocksWritten := make(chan struct{})
  77. wal := newByteBufferWAL(logger, NewWALEncoder(wr), int64(numBlocks), numBlocksWritten)
  78. // see wal.go#103
  79. wal.Save(EndHeightMessage{0})
  80. consensusState.wal = wal
  81. if err := consensusState.Start(); err != nil {
  82. return nil, errors.Wrap(err, "failed to start consensus state")
  83. }
  84. defer consensusState.Stop()
  85. select {
  86. case <-numBlocksWritten:
  87. wr.Flush()
  88. return b.Bytes(), nil
  89. case <-time.After(1 * time.Minute):
  90. wr.Flush()
  91. return b.Bytes(), fmt.Errorf("waited too long for tendermint to produce %d blocks (grep logs for `wal_generator`)", numBlocks)
  92. }
  93. }
  94. // f**ing long, but unique for each test
  95. func makePathname() string {
  96. // get path
  97. p, err := os.Getwd()
  98. if err != nil {
  99. panic(err)
  100. }
  101. // fmt.Println(p)
  102. sep := string(filepath.Separator)
  103. return strings.Replace(p, sep, "_", -1)
  104. }
  105. func randPort() int {
  106. // returns between base and base + spread
  107. base, spread := 20000, 20000
  108. return base + rand.Intn(spread)
  109. }
  110. func makeAddrs() (string, string, string) {
  111. start := randPort()
  112. return fmt.Sprintf("tcp://0.0.0.0:%d", start),
  113. fmt.Sprintf("tcp://0.0.0.0:%d", start+1),
  114. fmt.Sprintf("tcp://0.0.0.0:%d", start+2)
  115. }
  116. // getConfig returns a config for test cases
  117. func getConfig() *cfg.Config {
  118. pathname := makePathname()
  119. c := cfg.ResetTestRoot(fmt.Sprintf("%s_%d", pathname, cmn.RandInt()))
  120. // and we use random ports to run in parallel
  121. tm, rpc, grpc := makeAddrs()
  122. c.P2P.ListenAddress = tm
  123. c.RPC.ListenAddress = rpc
  124. c.RPC.GRPCListenAddress = grpc
  125. return c
  126. }
  127. // byteBufferWAL is a WAL which writes all msgs to a byte buffer. Writing stops
  128. // when the heightToStop is reached. Client will be notified via
  129. // signalWhenStopsTo channel.
  130. type byteBufferWAL struct {
  131. enc *WALEncoder
  132. stopped bool
  133. heightToStop int64
  134. signalWhenStopsTo chan<- struct{}
  135. logger log.Logger
  136. }
  137. // needed for determinism
  138. var fixedTime, _ = time.Parse(time.RFC3339, "2017-01-02T15:04:05Z")
  139. func newByteBufferWAL(logger log.Logger, enc *WALEncoder, nBlocks int64, signalStop chan<- struct{}) *byteBufferWAL {
  140. return &byteBufferWAL{
  141. enc: enc,
  142. heightToStop: nBlocks,
  143. signalWhenStopsTo: signalStop,
  144. logger: logger,
  145. }
  146. }
  147. // Save writes message to the internal buffer except when heightToStop is
  148. // reached, in which case it will signal the caller via signalWhenStopsTo and
  149. // skip writing.
  150. func (w *byteBufferWAL) Save(m WALMessage) {
  151. if w.stopped {
  152. w.logger.Debug("WAL already stopped. Not writing message", "msg", m)
  153. return
  154. }
  155. if endMsg, ok := m.(EndHeightMessage); ok {
  156. w.logger.Debug("WAL write end height message", "height", endMsg.Height, "stopHeight", w.heightToStop)
  157. if endMsg.Height == w.heightToStop {
  158. w.logger.Debug("Stopping WAL at height", "height", endMsg.Height)
  159. w.signalWhenStopsTo <- struct{}{}
  160. w.stopped = true
  161. return
  162. }
  163. }
  164. w.logger.Debug("WAL Write Message", "msg", m)
  165. err := w.enc.Encode(&TimedWALMessage{fixedTime, m})
  166. if err != nil {
  167. panic(fmt.Sprintf("failed to encode the msg %v", m))
  168. }
  169. }
  170. func (w *byteBufferWAL) Group() *auto.Group {
  171. panic("not implemented")
  172. }
  173. func (w *byteBufferWAL) SearchForEndHeight(height int64, options *WALSearchOptions) (gr *auto.GroupReader, found bool, err error) {
  174. return nil, false, nil
  175. }
  176. func (w *byteBufferWAL) Start() error { return nil }
  177. func (w *byteBufferWAL) Stop() error { return nil }
  178. func (w *byteBufferWAL) Wait() {}