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.

309 lines
12 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package consensus
  2. import (
  3. "fmt"
  4. "sync"
  5. "testing"
  6. "time"
  7. "github.com/tendermint/abci/example/dummy"
  8. "github.com/tendermint/tendermint/p2p"
  9. "github.com/tendermint/tendermint/types"
  10. "github.com/tendermint/tmlibs/events"
  11. )
  12. func init() {
  13. config = ResetConfig("consensus_reactor_test")
  14. }
  15. //----------------------------------------------
  16. // in-process testnets
  17. func startConsensusNet(t *testing.T, css []*ConsensusState, N int, subscribeEventRespond bool) ([]*ConsensusReactor, []chan interface{}) {
  18. reactors := make([]*ConsensusReactor, N)
  19. eventChans := make([]chan interface{}, N)
  20. logger := consensusLogger()
  21. for i := 0; i < N; i++ {
  22. reactors[i] = NewConsensusReactor(css[i], true) // so we dont start the consensus states
  23. reactors[i].SetLogger(logger.With("validator", i))
  24. eventSwitch := events.NewEventSwitch()
  25. eventSwitch.SetLogger(logger.With("module", "events", "validator", i))
  26. _, err := eventSwitch.Start()
  27. if err != nil {
  28. t.Fatalf("Failed to start switch: %v", err)
  29. }
  30. reactors[i].SetEventSwitch(eventSwitch)
  31. if subscribeEventRespond {
  32. eventChans[i] = subscribeToEventRespond(eventSwitch, "tester", types.EventStringNewBlock())
  33. } else {
  34. eventChans[i] = subscribeToEvent(eventSwitch, "tester", types.EventStringNewBlock(), 1)
  35. }
  36. }
  37. // make connected switches and start all reactors
  38. p2p.MakeConnectedSwitches(config.P2P, N, func(i int, s *p2p.Switch) *p2p.Switch {
  39. s.AddReactor("CONSENSUS", reactors[i])
  40. return s
  41. }, p2p.Connect2Switches)
  42. // now that everyone is connected, start the state machines
  43. // If we started the state machines before everyone was connected,
  44. // we'd block when the cs fires NewBlockEvent and the peers are trying to start their reactors
  45. for i := 0; i < N; i++ {
  46. s := reactors[i].conS.GetState()
  47. reactors[i].SwitchToConsensus(s)
  48. }
  49. return reactors, eventChans
  50. }
  51. func stopConsensusNet(reactors []*ConsensusReactor) {
  52. for _, r := range reactors {
  53. r.Switch.Stop()
  54. }
  55. }
  56. // Ensure a testnet makes blocks
  57. func TestReactor(t *testing.T) {
  58. N := 4
  59. css := randConsensusNet(N, "consensus_reactor_test", newMockTickerFunc(true), newCounter)
  60. reactors, eventChans := startConsensusNet(t, css, N, false)
  61. defer stopConsensusNet(reactors)
  62. // wait till everyone makes the first new block
  63. timeoutWaitGroup(t, N, func(wg *sync.WaitGroup, j int) {
  64. <-eventChans[j]
  65. wg.Done()
  66. }, css)
  67. }
  68. //-------------------------------------------------------------
  69. // ensure we can make blocks despite cycling a validator set
  70. func TestVotingPowerChange(t *testing.T) {
  71. nVals := 4
  72. css := randConsensusNet(nVals, "consensus_voting_power_changes_test", newMockTickerFunc(true), newPersistentDummy)
  73. reactors, eventChans := startConsensusNet(t, css, nVals, true)
  74. defer stopConsensusNet(reactors)
  75. // map of active validators
  76. activeVals := make(map[string]struct{})
  77. for i := 0; i < nVals; i++ {
  78. activeVals[string(css[i].privValidator.GetAddress())] = struct{}{}
  79. }
  80. // wait till everyone makes block 1
  81. timeoutWaitGroup(t, nVals, func(wg *sync.WaitGroup, j int) {
  82. <-eventChans[j]
  83. eventChans[j] <- struct{}{}
  84. wg.Done()
  85. }, css)
  86. //---------------------------------------------------------------------------
  87. t.Log("---------------------------- Testing changing the voting power of one validator a few times")
  88. val1PubKey := css[0].privValidator.(*types.PrivValidator).PubKey
  89. updateValidatorTx := dummy.MakeValSetChangeTx(val1PubKey.Bytes(), 25)
  90. previousTotalVotingPower := css[0].GetRoundState().LastValidators.TotalVotingPower()
  91. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css, updateValidatorTx)
  92. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  93. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  94. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  95. if css[0].GetRoundState().LastValidators.TotalVotingPower() == previousTotalVotingPower {
  96. t.Fatalf("expected voting power to change (before: %d, after: %d)", previousTotalVotingPower, css[0].GetRoundState().LastValidators.TotalVotingPower())
  97. }
  98. updateValidatorTx = dummy.MakeValSetChangeTx(val1PubKey.Bytes(), 2)
  99. previousTotalVotingPower = css[0].GetRoundState().LastValidators.TotalVotingPower()
  100. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css, updateValidatorTx)
  101. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  102. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  103. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  104. if css[0].GetRoundState().LastValidators.TotalVotingPower() == previousTotalVotingPower {
  105. t.Fatalf("expected voting power to change (before: %d, after: %d)", previousTotalVotingPower, css[0].GetRoundState().LastValidators.TotalVotingPower())
  106. }
  107. updateValidatorTx = dummy.MakeValSetChangeTx(val1PubKey.Bytes(), 100)
  108. previousTotalVotingPower = css[0].GetRoundState().LastValidators.TotalVotingPower()
  109. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css, updateValidatorTx)
  110. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  111. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  112. waitForAndValidateBlock(t, nVals, activeVals, eventChans, css)
  113. if css[0].GetRoundState().LastValidators.TotalVotingPower() == previousTotalVotingPower {
  114. t.Fatalf("expected voting power to change (before: %d, after: %d)", previousTotalVotingPower, css[0].GetRoundState().LastValidators.TotalVotingPower())
  115. }
  116. }
  117. func TestValidatorSetChanges(t *testing.T) {
  118. nPeers := 7
  119. nVals := 4
  120. css := randConsensusNetWithPeers(nVals, nPeers, "consensus_val_set_changes_test", newMockTickerFunc(true), newPersistentDummy)
  121. reactors, eventChans := startConsensusNet(t, css, nPeers, true)
  122. defer stopConsensusNet(reactors)
  123. // map of active validators
  124. activeVals := make(map[string]struct{})
  125. for i := 0; i < nVals; i++ {
  126. activeVals[string(css[i].privValidator.GetAddress())] = struct{}{}
  127. }
  128. // wait till everyone makes block 1
  129. timeoutWaitGroup(t, nPeers, func(wg *sync.WaitGroup, j int) {
  130. <-eventChans[j]
  131. eventChans[j] <- struct{}{}
  132. wg.Done()
  133. }, css)
  134. //---------------------------------------------------------------------------
  135. t.Log("---------------------------- Testing adding one validator")
  136. newValidatorPubKey1 := css[nVals].privValidator.(*types.PrivValidator).PubKey
  137. newValidatorTx1 := dummy.MakeValSetChangeTx(newValidatorPubKey1.Bytes(), uint64(testMinPower))
  138. // wait till everyone makes block 2
  139. // ensure the commit includes all validators
  140. // send newValTx to change vals in block 3
  141. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css, newValidatorTx1)
  142. // wait till everyone makes block 3.
  143. // it includes the commit for block 2, which is by the original validator set
  144. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  145. // wait till everyone makes block 4.
  146. // it includes the commit for block 3, which is by the original validator set
  147. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  148. // the commits for block 4 should be with the updated validator set
  149. activeVals[string(newValidatorPubKey1.Address())] = struct{}{}
  150. // wait till everyone makes block 5
  151. // it includes the commit for block 4, which should have the updated validator set
  152. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  153. //---------------------------------------------------------------------------
  154. t.Log("---------------------------- Testing changing the voting power of one validator")
  155. updateValidatorPubKey1 := css[nVals].privValidator.(*types.PrivValidator).PubKey
  156. updateValidatorTx1 := dummy.MakeValSetChangeTx(updateValidatorPubKey1.Bytes(), 25)
  157. previousTotalVotingPower := css[nVals].GetRoundState().LastValidators.TotalVotingPower()
  158. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css, updateValidatorTx1)
  159. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  160. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  161. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  162. if css[nVals].GetRoundState().LastValidators.TotalVotingPower() == previousTotalVotingPower {
  163. t.Errorf("expected voting power to change (before: %d, after: %d)", previousTotalVotingPower, css[nVals].GetRoundState().LastValidators.TotalVotingPower())
  164. }
  165. //---------------------------------------------------------------------------
  166. t.Log("---------------------------- Testing adding two validators at once")
  167. newValidatorPubKey2 := css[nVals+1].privValidator.(*types.PrivValidator).PubKey
  168. newValidatorTx2 := dummy.MakeValSetChangeTx(newValidatorPubKey2.Bytes(), uint64(testMinPower))
  169. newValidatorPubKey3 := css[nVals+2].privValidator.(*types.PrivValidator).PubKey
  170. newValidatorTx3 := dummy.MakeValSetChangeTx(newValidatorPubKey3.Bytes(), uint64(testMinPower))
  171. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css, newValidatorTx2, newValidatorTx3)
  172. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  173. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  174. activeVals[string(newValidatorPubKey2.Address())] = struct{}{}
  175. activeVals[string(newValidatorPubKey3.Address())] = struct{}{}
  176. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  177. //---------------------------------------------------------------------------
  178. t.Log("---------------------------- Testing removing two validators at once")
  179. removeValidatorTx2 := dummy.MakeValSetChangeTx(newValidatorPubKey2.Bytes(), 0)
  180. removeValidatorTx3 := dummy.MakeValSetChangeTx(newValidatorPubKey3.Bytes(), 0)
  181. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css, removeValidatorTx2, removeValidatorTx3)
  182. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  183. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  184. delete(activeVals, string(newValidatorPubKey2.Address()))
  185. delete(activeVals, string(newValidatorPubKey3.Address()))
  186. waitForAndValidateBlock(t, nPeers, activeVals, eventChans, css)
  187. }
  188. // Check we can make blocks with skip_timeout_commit=false
  189. func TestReactorWithTimeoutCommit(t *testing.T) {
  190. N := 4
  191. css := randConsensusNet(N, "consensus_reactor_with_timeout_commit_test", newMockTickerFunc(false), newCounter)
  192. // override default SkipTimeoutCommit == true for tests
  193. for i := 0; i < N; i++ {
  194. css[i].config.SkipTimeoutCommit = false
  195. }
  196. reactors, eventChans := startConsensusNet(t, css, N-1, false)
  197. defer stopConsensusNet(reactors)
  198. // wait till everyone makes the first new block
  199. timeoutWaitGroup(t, N-1, func(wg *sync.WaitGroup, j int) {
  200. <-eventChans[j]
  201. wg.Done()
  202. }, css)
  203. }
  204. func waitForAndValidateBlock(t *testing.T, n int, activeVals map[string]struct{}, eventChans []chan interface{}, css []*ConsensusState, txs ...[]byte) {
  205. timeoutWaitGroup(t, n, func(wg *sync.WaitGroup, j int) {
  206. newBlockI := <-eventChans[j]
  207. newBlock := newBlockI.(types.TMEventData).Unwrap().(types.EventDataNewBlock).Block
  208. t.Logf("[WARN] Got block height=%v validator=%v", newBlock.Height, j)
  209. err := validateBlock(newBlock, activeVals)
  210. if err != nil {
  211. t.Fatal(err)
  212. }
  213. for _, tx := range txs {
  214. css[j].mempool.CheckTx(tx, nil)
  215. }
  216. eventChans[j] <- struct{}{}
  217. wg.Done()
  218. }, css)
  219. }
  220. // expects high synchrony!
  221. func validateBlock(block *types.Block, activeVals map[string]struct{}) error {
  222. if block.LastCommit.Size() != len(activeVals) {
  223. return fmt.Errorf("Commit size doesn't match number of active validators. Got %d, expected %d", block.LastCommit.Size(), len(activeVals))
  224. }
  225. for _, vote := range block.LastCommit.Precommits {
  226. if _, ok := activeVals[string(vote.ValidatorAddress)]; !ok {
  227. return fmt.Errorf("Found vote for unactive validator %X", vote.ValidatorAddress)
  228. }
  229. }
  230. return nil
  231. }
  232. func timeoutWaitGroup(t *testing.T, n int, f func(*sync.WaitGroup, int), css []*ConsensusState) {
  233. wg := new(sync.WaitGroup)
  234. wg.Add(n)
  235. for i := 0; i < n; i++ {
  236. go f(wg, i)
  237. }
  238. done := make(chan struct{})
  239. go func() {
  240. wg.Wait()
  241. close(done)
  242. }()
  243. select {
  244. case <-done:
  245. case <-time.After(time.Second * 10):
  246. for i, cs := range css {
  247. fmt.Println("#################")
  248. fmt.Println("Validator", i)
  249. fmt.Println(cs.GetRoundState())
  250. fmt.Println("")
  251. }
  252. panic("Timed out waiting for all validators to commit a block")
  253. }
  254. }