add test cases

plugin/consensus/dpos/chain33.test.toml

@@ -100,15 +100,15 @@ timeoutWaitNotify=2000
 createEmptyBlocks=false
 createEmptyBlocksInterval=0
 validatorNodes=["127.0.0.1:46656"]
-delegateNum=3
+delegateNum=1
 blockInterval=3
 continueBlockNum=12
-isValidator=false
+isValidator=true
 rpcAddr="http://localhost:9801"
 #shuffleType为1表示使用固定出块顺序，为2表示使用vrf信息进行出块顺序洗牌
-shuffleType=2
+shuffleType=1
 #是否更新topN，如果为true，根据下面几个配置项定期更新topN节点;如果为false，则一直使用初始配置的节点，不关注投票结果
-whetherUpdateTopN=true
+whetherUpdateTopN=false
 blockNumToUpdateDelegate=20000
 registTopNHeightLimit=100
 updateTopNHeightLimit=200

plugin/consensus/dpos/dpos.go

@@ -273,8 +273,12 @@ func (client *Client) StartConsensus() {
 	//进入共识前先同步到最大高度
 	hint := time.NewTicker(5 * time.Second)
 	beg := time.Now()
+	block, err := client.RequestLastBlock()
+	if err != nil {
+		panic(err)
+	}
 OuterLoop:
-	for !DebugCatchup {
+	for !DebugCatchup && block != nil{
 		select {
 		case <-hint.C:
 			dposlog.Info("Still catching up max height......", "cost", time.Since(beg))
@@ -303,7 +307,7 @@ OuterLoop:
 	}
 	valMgr = valMgrTmp.Copy()
 	dposlog.Debug("Load Validator Manager finish", "state", valMgr)
-	block, err := client.RequestLastBlock()
+	block, err = client.RequestLastBlock()
 	if err != nil {
 		panic(err)
 	}
@@ -351,6 +355,7 @@ OuterLoop:
 		client.isDelegator = true
 	} else {
 		dposlog.Info("This node is not a delegator")
+		dposlog.Info("StartConsensus", "privValidator addr", hex.EncodeToString(client.privValidator.GetAddress()))
 	}
 
 	// Make ConsensusReactor
@@ -380,7 +385,7 @@ OuterLoop:
 func printValidators(set *ttypes.ValidatorSet) string {
 	result := "Validators:["
 	for _, v := range set.Validators {
-		result = fmt.Sprintf("%s%s,", result, hex.EncodeToString(v.PubKey))
+		result = fmt.Sprintf("%s%s|%s,", result, hex.EncodeToString(v.PubKey), hex.EncodeToString(v.Address))
 	}
 
 	result += "]"

plugin/consensus/dpos/dpos_test.go

+// Copyright Fuzamei Corp. 2018 All Rights Reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package dpos
+
+import (
+	"context"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"flag"
+	"fmt"
+	"github.com/33cn/chain33/common/crypto"
+	"math/rand"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/33cn/chain33/blockchain"
+	"github.com/33cn/chain33/common/address"
+	"github.com/33cn/chain33/common/limits"
+	"github.com/33cn/chain33/common/log"
+	"github.com/33cn/chain33/executor"
+	"github.com/33cn/chain33/mempool"
+	"github.com/33cn/chain33/p2p"
+	"github.com/33cn/chain33/queue"
+	"github.com/33cn/chain33/rpc"
+	"github.com/33cn/chain33/store"
+	"github.com/33cn/chain33/types"
+	ttypes "github.com/33cn/plugin/plugin/consensus/dpos/types"
+	pty "github.com/33cn/plugin/plugin/dapp/norm/types"
+	"google.golang.org/grpc"
+
+	_ "github.com/33cn/chain33/system"
+	_ "github.com/33cn/plugin/plugin/dapp/init"
+	_ "github.com/33cn/plugin/plugin/store/init"
+)
+
+var (
+	random    *rand.Rand
+	loopCount = 10
+	conn      *grpc.ClientConn
+	c         types.Chain33Client
+)
+const fee = 1e6
+
+
+func init() {
+	err := limits.SetLimits()
+	if err != nil {
+		panic(err)
+	}
+	random = rand.New(rand.NewSource(types.Now().UnixNano()))
+	log.SetLogLevel("info")
+}
+func TestTendermintPerf(t *testing.T) {
+	TendermintPerf()
+	fmt.Println("=======start clear test data!=======")
+	clearTestData()
+}
+
+func TendermintPerf() {
+	q, chain, s, mem, exec, cs, p2p := initEnvTendermint()
+	defer chain.Close()
+	defer mem.Close()
+	defer exec.Close()
+	defer s.Close()
+	defer q.Close()
+	defer cs.Close()
+	defer p2p.Close()
+	err := createConn()
+	for err != nil {
+		err = createConn()
+	}
+	time.Sleep(10 * time.Second)
+	for i := 0; i < loopCount; i++ {
+		NormPut()
+		time.Sleep(time.Second)
+	}
+	time.Sleep(10 * time.Second)
+}
+
+func initEnvTendermint() (queue.Queue, *blockchain.BlockChain, queue.Module, queue.Module, *executor.Executor, queue.Module, queue.Module) {
+	var q = queue.New("channel")
+	flag.Parse()
+	cfg, sub := types.InitCfg("chain33.test.toml")
+	types.Init(cfg.Title, cfg)
+	chain := blockchain.New(cfg.BlockChain)
+	chain.SetQueueClient(q.Client())
+
+	exec := executor.New(cfg.Exec, sub.Exec)
+	exec.SetQueueClient(q.Client())
+	types.SetMinFee(0)
+	s := store.New(cfg.Store, sub.Store)
+	s.SetQueueClient(q.Client())
+
+	cs := New(cfg.Consensus, sub.Consensus["dpos"])
+	cs.SetQueueClient(q.Client())
+
+	mem := mempool.New(cfg.Mempool, nil)
+	mem.SetQueueClient(q.Client())
+	network := p2p.New(cfg.P2P)
+
+	network.SetQueueClient(q.Client())
+
+	rpc.InitCfg(cfg.RPC)
+	gapi := rpc.NewGRpcServer(q.Client(), nil)
+	go gapi.Listen()
+	return q, chain, s, mem, exec, cs, network
+}
+
+func createConn() error {
+	var err error
+	url := "127.0.0.1:8802"
+	fmt.Println("grpc url:", url)
+	conn, err = grpc.Dial(url, grpc.WithInsecure())
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		return err
+	}
+	c = types.NewChain33Client(conn)
+	//r = rand.New(rand.NewSource(types.Now().UnixNano()))
+	return nil
+}
+
+func generateKey(i, valI int) string {
+	key := make([]byte, valI)
+	binary.PutUvarint(key[:10], uint64(valI))
+	binary.PutUvarint(key[12:24], uint64(i))
+	if _, err := rand.Read(key[24:]); err != nil {
+		os.Exit(1)
+	}
+	return string(key)
+}
+
+func generateValue(i, valI int) string {
+	value := make([]byte, valI)
+	binary.PutUvarint(value[:16], uint64(i))
+	binary.PutUvarint(value[32:128], uint64(i))
+	if _, err := rand.Read(value[128:]); err != nil {
+		os.Exit(1)
+	}
+	return string(value)
+}
+
+func getprivkey(key string) crypto.PrivKey {
+	bkey, err := hex.DecodeString(key)
+	if err != nil {
+		panic(err)
+	}
+	priv, err := ttypes.ConsensusCrypto.PrivKeyFromBytes(bkey)
+	if err != nil {
+		panic(err)
+	}
+	return priv
+}
+
+func prepareTxList() *types.Transaction {
+	var key string
+	var value string
+	var i int
+
+	key = generateKey(i, 32)
+	value = generateValue(i, 180)
+
+	nput := &pty.NormAction_Nput{Nput: &pty.NormPut{Key: []byte(key), Value: []byte(value)}}
+	action := &pty.NormAction{Value: nput, Ty: pty.NormActionPut}
+	tx := &types.Transaction{Execer: []byte("norm"), Payload: types.Encode(action), Fee: fee}
+	tx.To = address.ExecAddress("norm")
+	tx.Nonce = random.Int63()
+	tx.Sign(types.SECP256K1, getprivkey("CC38546E9E659D15E6B4893F0AB32A06D103931A8230B0BDE71459D2B27D6944"))
+	return tx
+}
+
+func clearTestData() {
+	err := os.RemoveAll("datadir")
+	if err != nil {
+		fmt.Println("delete datadir have a err:", err.Error())
+	}
+	fmt.Println("test data clear successfully!")
+}
+
+func NormPut() {
+	tx := prepareTxList()
+
+	reply, err := c.SendTransaction(context.Background(), tx)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		return
+	}
+	if !reply.IsOk {
+		fmt.Fprintln(os.Stderr, errors.New(string(reply.GetMsg())))
+		return
+	}
+}

plugin/consensus/dpos/priv_validator.json

-{"address":"02A13174B92727C4902DB099E51A3339F48BD45E","pub_key":{"type":"ed25519","data":"220ACBE680DF2473A0CB48987A00FCC1812F106A7390BE6B8E2D31122C992A19"},"priv_key":{"type":"ed25519","data":"B3DC4C0725884EBB7264B92F1D8D37584A64ADE1799D997EC64B4FE3973E08DE220ACBE680DF2473A0CB48987A00FCC1812F106A7390BE6B8E2D31122C992A19"}}
\ No newline at end of file
+{"address":"2B226E6603E52C94715BA4E92080EEF236292E33","pub_key":{"type":"secp256k1","data":"03EF0E1D3112CF571743A3318125EDE2E52A4EB904BCBAA4B1F75020C2846A7EB4"},"last_height":1679,"last_round":0,"last_step":3,"last_signature":{"type":"secp256k1","data":"37892A916D6E487ADF90F9E88FE37024597677B6C6FED47444AD582F74144B3D6E4B364EAF16AF03A4E42827B6D3C86415D734A5A6CCA92E114B23EB9265AF09"},"last_signbytes":"7B22636861696E5F6964223A22636861696E33332D5A326367466A222C22766F7465223A7B22626C6F636B5F6964223A7B2268617368223A224F6A657975396B2B4149426A6E4859456739584765356A7A462B673D222C227061727473223A7B2268617368223A6E756C6C2C22746F74616C223A307D7D2C22686569676874223A313637392C22726F756E64223A302C2274696D657374616D70223A22323031382D30382D33315430373A35313A34332E3935395A222C2274797065223A327D7D","priv_key":{"type":"secp256k1","data":"5A6A14DA6F5A42835E529D75D87CC8904544F59EEE5387A37D87EEAD194D7EB2"}}
\ No newline at end of file

plugin/consensus/dpos/state_machine.go

@@ -383,7 +383,7 @@ func (init *InitState) timeOut(cs *ConsensusState) {
 	//if available noes  < 2/3, don't change the state to voting.
 	connections := cs.client.node.peerSet.Size()
 	validators := cs.validatorMgr.Validators.Size()
-	if connections == 0 || connections < (validators*2/3-1) {
+	if dposDelegateNum > 1 && (connections == 0 || connections < (validators*2/3-1)) {
 		dposlog.Error("InitState timeout but available nodes less than 2/3,waiting for more connections", "connections", connections, "validators", validators)
 		cs.ClearVotes()
 
@@ -460,6 +460,31 @@ type VotingState struct {
 }
 
 func (voting *VotingState) timeOut(cs *ConsensusState) {
+	//如果是测试场景，只有一个节点，也需要状态机能运转下去
+	if dposDelegateNum == 1 {
+		result, voteItem := cs.CheckVotes()
+
+		if result == voteSuccess {
+			dposlog.Info("VotingState get 2/3 result", "final vote:", printVoteItem(voteItem))
+			dposlog.Info("VotingState change state to VotedState")
+			//切换状态
+			cs.SetState(VotedStateObj)
+			dposlog.Info("Change state because of check votes successfully.", "from", "VotingState", "to", "VotedState")
+
+			cs.SetCurrentVote(voteItem)
+
+			//检查最终投票是否与自己的投票一致，如果不一致，需要更新本地的信息，保证各节点共识结果执行一致。
+			if !bytes.Equal(cs.myVote.VoteItem.VoteID, voteItem.VoteID) {
+				if !cs.validatorMgr.UpdateFromVoteItem(voteItem) {
+					panic("This node's validators are not the same with final vote, please check")
+				}
+			}
+			//1s后检查是否出块，是否需要重新投票
+			cs.scheduleDPosTimeout(time.Millisecond*500, VotedStateType)
+		}
+		return
+	}
+
 	dposlog.Info("VotingState timeout but don't get an agreement. change state to InitState")
 
 	//清理掉之前的选票记录，从初始状态重新开始
@@ -548,6 +573,7 @@ func (voted *VotedState) timeOut(cs *ConsensusState) {
 	block := cs.client.GetCurrentBlock()
 	task := DecideTaskByTime(now)
 
+	dposlog.Info("address info", "privValidatorAddr", hex.EncodeToString(cs.privValidator.GetAddress()), "VotedNodeAddress",hex.EncodeToString(cs.currentVote.VotedNodeAddress))
 	if bytes.Equal(cs.privValidator.GetAddress(), cs.currentVote.VotedNodeAddress) {
 		//当前节点为出块节点