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

9 years ago
  1. package consensus
  2. import (
  3. "bytes"
  4. "fmt"
  5. "sort"
  6. "sync"
  7. "testing"
  8. "time"
  9. cfg "github.com/tendermint/go-config"
  10. dbm "github.com/tendermint/go-db"
  11. bc "github.com/tendermint/tendermint/blockchain"
  12. mempl "github.com/tendermint/tendermint/mempool"
  13. sm "github.com/tendermint/tendermint/state"
  14. "github.com/tendermint/tendermint/types"
  15. tmspcli "github.com/tendermint/tmsp/client"
  16. tmsp "github.com/tendermint/tmsp/types"
  17. "github.com/tendermint/tmsp/example/counter"
  18. "github.com/tendermint/tmsp/example/dummy"
  19. )
  20. var config cfg.Config // NOTE: must be reset for each _test.go file
  21. var ensureTimeout = time.Duration(2)
  22. type validatorStub struct {
  23. Height int
  24. Round int
  25. *types.PrivValidator
  26. }
  27. func NewValidatorStub(privValidator *types.PrivValidator) *validatorStub {
  28. return &validatorStub{
  29. PrivValidator: privValidator,
  30. }
  31. }
  32. func (vs *validatorStub) signVote(voteType byte, hash []byte, header types.PartSetHeader) (*types.Vote, error) {
  33. vote := &types.Vote{
  34. Height: vs.Height,
  35. Round: vs.Round,
  36. Type: voteType,
  37. BlockHash: hash,
  38. BlockPartsHeader: header,
  39. }
  40. err := vs.PrivValidator.SignVote(config.GetString("chain_id"), vote)
  41. return vote, err
  42. }
  43. // convenienve function for testing
  44. func signVote(vs *validatorStub, voteType byte, hash []byte, header types.PartSetHeader) *types.Vote {
  45. v, err := vs.signVote(voteType, hash, header)
  46. if err != nil {
  47. panic(fmt.Errorf("failed to sign vote: %v", err))
  48. }
  49. return v
  50. }
  51. // create proposal block from cs1 but sign it with vs
  52. func decideProposal(cs1 *ConsensusState, cs2 *validatorStub, height, round int) (proposal *types.Proposal, block *types.Block) {
  53. block, blockParts := cs1.createProposalBlock()
  54. if block == nil { // on error
  55. panic("error creating proposal block")
  56. }
  57. // Make proposal
  58. proposal = types.NewProposal(height, round, blockParts.Header(), cs1.Votes.POLRound())
  59. if err := cs2.SignProposal(config.GetString("chain_id"), proposal); err != nil {
  60. panic(err)
  61. }
  62. return
  63. }
  64. //-------------------------------------------------------------------------------
  65. // utils
  66. /*
  67. func nilRound(t *testing.T, cs1 *ConsensusState, vss ...*validatorStub) {
  68. cs1.mtx.Lock()
  69. height, round := cs1.Height, cs1.Round
  70. cs1.mtx.Unlock()
  71. waitFor(t, cs1, height, round, RoundStepPrevote)
  72. signAddVoteToFromMany(types.VoteTypePrevote, cs1, nil, cs1.ProposalBlockParts.Header(), vss...)
  73. waitFor(t, cs1, height, round, RoundStepPrecommit)
  74. signAddVoteToFromMany(types.VoteTypePrecommit, cs1, nil, cs1.ProposalBlockParts.Header(), vss...)
  75. waitFor(t, cs1, height, round+1, RoundStepNewRound)
  76. }
  77. */
  78. // NOTE: this switches the propser as far as `perspectiveOf` is concerned,
  79. // but for simplicity we return a block it generated.
  80. func changeProposer(t *testing.T, perspectiveOf *ConsensusState, newProposer *validatorStub) *types.Block {
  81. _, v1 := perspectiveOf.Validators.GetByAddress(perspectiveOf.privValidator.Address)
  82. v1.Accum, v1.VotingPower = 0, 0
  83. if updated := perspectiveOf.Validators.Update(v1); !updated {
  84. panic("failed to update validator")
  85. }
  86. _, v2 := perspectiveOf.Validators.GetByAddress(newProposer.Address)
  87. v2.Accum, v2.VotingPower = 100, 100
  88. if updated := perspectiveOf.Validators.Update(v2); !updated {
  89. panic("failed to update validator")
  90. }
  91. // make the proposal
  92. propBlock, _ := perspectiveOf.createProposalBlock()
  93. if propBlock == nil {
  94. panic("Failed to create proposal block with cs2")
  95. }
  96. return propBlock
  97. }
  98. func fixVotingPower(t *testing.T, cs1 *ConsensusState, addr2 []byte) {
  99. _, v1 := cs1.Validators.GetByAddress(cs1.privValidator.Address)
  100. _, v2 := cs1.Validators.GetByAddress(addr2)
  101. v1.Accum, v1.VotingPower = v2.Accum, v2.VotingPower
  102. if updated := cs1.Validators.Update(v1); !updated {
  103. panic("failed to update validator")
  104. }
  105. }
  106. func addVoteToFromMany(to *ConsensusState, votes []*types.Vote, froms ...*validatorStub) {
  107. if len(votes) != len(froms) {
  108. panic("len(votes) and len(froms) must match")
  109. }
  110. for i, from := range froms {
  111. addVoteToFrom(to, from, votes[i])
  112. }
  113. }
  114. func addVoteToFrom(to *ConsensusState, from *validatorStub, vote *types.Vote) {
  115. to.mtx.Lock() // NOTE: wont need this when the vote comes with the index!
  116. valIndex, _ := to.Validators.GetByAddress(from.PrivValidator.Address)
  117. to.mtx.Unlock()
  118. to.peerMsgQueue <- msgInfo{Msg: &VoteMessage{valIndex, vote}}
  119. // added, err := to.TryAddVote(valIndex, vote, "")
  120. /*
  121. if _, ok := err.(*types.ErrVoteConflictingSignature); ok {
  122. // let it fly
  123. } else if !added {
  124. fmt.Println("to, from, vote:", to.Height, from.Height, vote.Height)
  125. panic(fmt.Sprintln("Failed to add vote. Err:", err))
  126. } else if err != nil {
  127. panic(fmt.Sprintln("Failed to add vote:", err))
  128. }*/
  129. }
  130. func signVoteMany(voteType byte, hash []byte, header types.PartSetHeader, vss ...*validatorStub) []*types.Vote {
  131. votes := make([]*types.Vote, len(vss))
  132. for i, vs := range vss {
  133. votes[i] = signVote(vs, voteType, hash, header)
  134. }
  135. return votes
  136. }
  137. // add vote to one cs from another
  138. // if voteCh is not nil, read all votes
  139. func signAddVoteToFromMany(voteType byte, to *ConsensusState, hash []byte, header types.PartSetHeader, voteCh chan interface{}, froms ...*validatorStub) {
  140. var wg chan struct{} // when done reading all votes
  141. if voteCh != nil {
  142. wg = readVotes(voteCh, len(froms))
  143. }
  144. for _, from := range froms {
  145. vote := signVote(from, voteType, hash, header)
  146. addVoteToFrom(to, from, vote)
  147. }
  148. if voteCh != nil {
  149. <-wg
  150. }
  151. }
  152. func signAddVoteToFrom(voteType byte, to *ConsensusState, from *validatorStub, hash []byte, header types.PartSetHeader, voteCh chan interface{}) *types.Vote {
  153. var wg chan struct{} // when done reading all votes
  154. if voteCh != nil {
  155. wg = readVotes(voteCh, 1)
  156. }
  157. vote := signVote(from, voteType, hash, header)
  158. addVoteToFrom(to, from, vote)
  159. if voteCh != nil {
  160. <-wg
  161. }
  162. return vote
  163. }
  164. func ensureNoNewStep(stepCh chan interface{}) {
  165. timeout := time.NewTicker(ensureTimeout * time.Second)
  166. select {
  167. case <-timeout.C:
  168. break
  169. case <-stepCh:
  170. panic("We should be stuck waiting for more votes, not moving to the next step")
  171. }
  172. }
  173. /*
  174. func ensureNoNewStep(t *testing.T, cs *ConsensusState) {
  175. timeout := time.NewTicker(ensureTimeout * time.Second)
  176. select {
  177. case <-timeout.C:
  178. break
  179. case <-cs.NewStepCh():
  180. panic("We should be stuck waiting for more votes, not moving to the next step")
  181. }
  182. }
  183. func ensureNewStep(t *testing.T, cs *ConsensusState) *RoundState {
  184. timeout := time.NewTicker(ensureTimeout * time.Second)
  185. select {
  186. case <-timeout.C:
  187. panic("We should have gone to the next step, not be stuck waiting")
  188. case rs := <-cs.NewStepCh():
  189. return rs
  190. }
  191. }
  192. func waitFor(t *testing.T, cs *ConsensusState, height int, round int, step RoundStepType) {
  193. for {
  194. rs := ensureNewStep(t, cs)
  195. if CompareHRS(rs.Height, rs.Round, rs.Step, height, round, step) < 0 {
  196. continue
  197. } else {
  198. break
  199. }
  200. }
  201. }
  202. */
  203. func incrementHeight(vss ...*validatorStub) {
  204. for _, vs := range vss {
  205. vs.Height += 1
  206. }
  207. }
  208. func incrementRound(vss ...*validatorStub) {
  209. for _, vs := range vss {
  210. vs.Round += 1
  211. }
  212. }
  213. func validatePrevote(t *testing.T, cs *ConsensusState, round int, privVal *validatorStub, blockHash []byte) {
  214. prevotes := cs.Votes.Prevotes(round)
  215. var vote *types.Vote
  216. if vote = prevotes.GetByAddress(privVal.Address); vote == nil {
  217. panic("Failed to find prevote from validator")
  218. }
  219. if blockHash == nil {
  220. if vote.BlockHash != nil {
  221. panic(fmt.Sprintf("Expected prevote to be for nil, got %X", vote.BlockHash))
  222. }
  223. } else {
  224. if !bytes.Equal(vote.BlockHash, blockHash) {
  225. panic(fmt.Sprintf("Expected prevote to be for %X, got %X", blockHash, vote.BlockHash))
  226. }
  227. }
  228. }
  229. func validateLastPrecommit(t *testing.T, cs *ConsensusState, privVal *validatorStub, blockHash []byte) {
  230. votes := cs.LastCommit
  231. var vote *types.Vote
  232. if vote = votes.GetByAddress(privVal.Address); vote == nil {
  233. panic("Failed to find precommit from validator")
  234. }
  235. if !bytes.Equal(vote.BlockHash, blockHash) {
  236. panic(fmt.Sprintf("Expected precommit to be for %X, got %X", blockHash, vote.BlockHash))
  237. }
  238. }
  239. func validatePrecommit(t *testing.T, cs *ConsensusState, thisRound, lockRound int, privVal *validatorStub, votedBlockHash, lockedBlockHash []byte) {
  240. precommits := cs.Votes.Precommits(thisRound)
  241. var vote *types.Vote
  242. if vote = precommits.GetByAddress(privVal.Address); vote == nil {
  243. panic("Failed to find precommit from validator")
  244. }
  245. if votedBlockHash == nil {
  246. if vote.BlockHash != nil {
  247. panic("Expected precommit to be for nil")
  248. }
  249. } else {
  250. if !bytes.Equal(vote.BlockHash, votedBlockHash) {
  251. panic("Expected precommit to be for proposal block")
  252. }
  253. }
  254. if lockedBlockHash == nil {
  255. if cs.LockedRound != lockRound || cs.LockedBlock != nil {
  256. panic(fmt.Sprintf("Expected to be locked on nil at round %d. Got locked at round %d with block %v", lockRound, cs.LockedRound, cs.LockedBlock))
  257. }
  258. } else {
  259. if cs.LockedRound != lockRound || !bytes.Equal(cs.LockedBlock.Hash(), lockedBlockHash) {
  260. panic(fmt.Sprintf("Expected block to be locked on round %d, got %d. Got locked block %X, expected %X", lockRound, cs.LockedRound, cs.LockedBlock.Hash(), lockedBlockHash))
  261. }
  262. }
  263. }
  264. func validatePrevoteAndPrecommit(t *testing.T, cs *ConsensusState, thisRound, lockRound int, privVal *validatorStub, votedBlockHash, lockedBlockHash []byte) {
  265. // verify the prevote
  266. validatePrevote(t, cs, thisRound, privVal, votedBlockHash)
  267. // verify precommit
  268. cs.mtx.Lock()
  269. validatePrecommit(t, cs, thisRound, lockRound, privVal, votedBlockHash, lockedBlockHash)
  270. cs.mtx.Unlock()
  271. }
  272. func fixedConsensusState() *ConsensusState {
  273. stateDB := dbm.NewMemDB()
  274. state := sm.MakeGenesisStateFromFile(stateDB, config.GetString("genesis_file"))
  275. privValidatorFile := config.GetString("priv_validator_file")
  276. privValidator := types.LoadOrGenPrivValidator(privValidatorFile)
  277. privValidator.Reset()
  278. cs := newConsensusState(state, privValidator, counter.NewCounterApplication(true))
  279. return cs
  280. }
  281. func fixedConsensusStateDummy() *ConsensusState {
  282. stateDB := dbm.NewMemDB()
  283. state := sm.MakeGenesisStateFromFile(stateDB, config.GetString("genesis_file"))
  284. privValidatorFile := config.GetString("priv_validator_file")
  285. privValidator := types.LoadOrGenPrivValidator(privValidatorFile)
  286. privValidator.Reset()
  287. cs := newConsensusState(state, privValidator, dummy.NewDummyApplication())
  288. return cs
  289. }
  290. func newConsensusState(state *sm.State, pv *types.PrivValidator, app tmsp.Application) *ConsensusState {
  291. // Get BlockStore
  292. blockDB := dbm.NewMemDB()
  293. blockStore := bc.NewBlockStore(blockDB)
  294. // one for mempool, one for consensus
  295. mtx := new(sync.Mutex)
  296. proxyAppConnMem := tmspcli.NewLocalClient(mtx, app)
  297. proxyAppConnCon := tmspcli.NewLocalClient(mtx, app)
  298. // Make Mempool
  299. mempool := mempl.NewMempool(config, proxyAppConnMem)
  300. // Make ConsensusReactor
  301. cs := NewConsensusState(config, state, proxyAppConnCon, blockStore, mempool)
  302. cs.SetPrivValidator(pv)
  303. evsw := types.NewEventSwitch()
  304. cs.SetEventSwitch(evsw)
  305. evsw.Start()
  306. return cs
  307. }
  308. func randConsensusState(nValidators int) (*ConsensusState, []*validatorStub) {
  309. // Get State
  310. state, privVals := randGenesisState(nValidators, false, 10)
  311. vss := make([]*validatorStub, nValidators)
  312. cs := newConsensusState(state, privVals[0], counter.NewCounterApplication(true))
  313. for i := 0; i < nValidators; i++ {
  314. vss[i] = NewValidatorStub(privVals[i])
  315. }
  316. // since cs1 starts at 1
  317. incrementHeight(vss[1:]...)
  318. return cs, vss
  319. }
  320. func subscribeToVoter(cs *ConsensusState, addr []byte) chan interface{} {
  321. voteCh0 := subscribeToEvent(cs.evsw, "tester", types.EventStringVote(), 1)
  322. voteCh := make(chan interface{})
  323. go func() {
  324. for {
  325. v := <-voteCh0
  326. vote := v.(types.EventDataVote)
  327. // we only fire for our own votes
  328. if bytes.Equal(addr, vote.Address) {
  329. voteCh <- v
  330. }
  331. }
  332. }()
  333. return voteCh
  334. }
  335. func readVotes(ch chan interface{}, reads int) chan struct{} {
  336. wg := make(chan struct{})
  337. go func() {
  338. for i := 0; i < reads; i++ {
  339. <-ch // read the precommit event
  340. }
  341. close(wg)
  342. }()
  343. return wg
  344. }
  345. func randGenesisState(numValidators int, randPower bool, minPower int64) (*sm.State, []*types.PrivValidator) {
  346. db := dbm.NewMemDB()
  347. genDoc, privValidators := randGenesisDoc(numValidators, randPower, minPower)
  348. s0 := sm.MakeGenesisState(db, genDoc)
  349. s0.Save()
  350. return s0, privValidators
  351. }
  352. func randGenesisDoc(numValidators int, randPower bool, minPower int64) (*types.GenesisDoc, []*types.PrivValidator) {
  353. validators := make([]types.GenesisValidator, numValidators)
  354. privValidators := make([]*types.PrivValidator, numValidators)
  355. for i := 0; i < numValidators; i++ {
  356. val, privVal := types.RandValidator(randPower, minPower)
  357. validators[i] = types.GenesisValidator{
  358. PubKey: val.PubKey,
  359. Amount: val.VotingPower,
  360. }
  361. privValidators[i] = privVal
  362. }
  363. sort.Sort(types.PrivValidatorsByAddress(privValidators))
  364. return &types.GenesisDoc{
  365. GenesisTime: time.Now(),
  366. ChainID: config.GetString("chain_id"),
  367. Validators: validators,
  368. }, privValidators
  369. }
  370. func startTestRound(cs *ConsensusState, height, round int) {
  371. cs.enterNewRound(height, round)
  372. cs.startRoutines(0)
  373. }