routine bls cmt msg process

plugin/consensus/para/para.go

@@ -18,7 +18,6 @@ import (
 	"time"
 
 	"github.com/33cn/chain33/client/api"
-	"github.com/33cn/chain33/common"
 	"github.com/33cn/chain33/common/crypto"
 	"github.com/33cn/chain33/common/merkle"
 	"github.com/33cn/chain33/queue"
@@ -333,10 +332,7 @@ func (client *client) ProcEvent(msg *queue.Message) bool {
 			plog.Info("paracross ProcEvent from", "from", req.GetFrom(), "topic:", req.GetTopic(), "ty", sub.GetTy())
 			switch sub.GetTy() {
 			case P2pSubCommitTx:
-				err := client.blsSignCli.rcvCommitTx(sub.GetCommitTx())
-				if err != nil {
-					plog.Error("paracross ProcEvent commit tx", "err", err, "txhash", common.ToHex(sub.GetCommitTx().Hash()), "from", sub.GetCommitTx().From())
-				}
+				go client.blsSignCli.rcvCommitTx(sub.GetCommitTx())
 			case P2pSubLeaderSyncMsg:
 				err := client.blsSignCli.rcvLeaderSyncTx(sub.GetSyncMsg())
 				if err != nil {

plugin/consensus/para/parablssign.go

@@ -231,6 +231,7 @@ func (b *blsClient) sendAggregateTx(nodes []string) error {
 func (b *blsClient) rcvCommitTx(tx *types.Transaction) error {
 	if !b.isValidNodes(tx.From()) {
 		b.updatePeers(tx.From(), false)
+		plog.Error("rcvCommitTx is not valid node", "addr", tx.From())
 		return pt.ErrParaNodeAddrNotExisted
 	}
 
@@ -238,6 +239,7 @@ func (b *blsClient) rcvCommitTx(tx *types.Transaction) error {
 	if count := tx.GetGroupCount(); count > 0 {
 		group, err := tx.GetTxGroup()
 		if err != nil {
+			plog.Error("rcvCommitTx GetTxGroup ", "err", err)
 			return errors.Wrap(err, "GetTxGroup")
 		}
 		txs = group.Txs
@@ -245,9 +247,13 @@ func (b *blsClient) rcvCommitTx(tx *types.Transaction) error {
 
 	commits, err := b.checkCommitTx(txs)
 	if err != nil {
+		plog.Error("rcvCommitTx checkCommitTx ", "err", err)
 		return errors.Wrap(err, "checkCommitTx")
 	}
 	b.updatePeers(tx.From(), true)
+	if len(commits) > 0 {
+		plog.Debug("rcvCommitTx tx", "addr", tx.From(), "height", commits[0].Status.Height)
+	}
 
 	b.rcvCommitTxCh <- commits
 	return nil
@@ -415,6 +421,9 @@ func aggregateSigns(signs [][]byte) (*g2pubs.Signature, error) {
 }
 
 func (b *blsClient) updatePeers(id string, add bool) {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
+
 	if _, ok := b.peers[id]; ok {
 		if !add {
 			delete(b.peers, id)
@@ -451,6 +460,19 @@ func getBlsPriKey(key []byte) *g2pubs.SecretKey {
 
 }
 
+//transfer secp Private key to bls pub key
+func secpPrikey2BlsPub(key string) (string, error) {
+	secpPrkKey, err := getSecpPriKey(key)
+	if err != nil {
+		plog.Error("getSecpPriKey", "err", err)
+		return "", err
+	}
+	blsPriKey := getBlsPriKey(secpPrkKey.Bytes())
+	blsPubKey := g2pubs.PrivToPub(blsPriKey)
+	serial := blsPubKey.Serialize()
+	return common.ToHex(serial[:]), nil
+}
+
 func (b *blsClient) blsSign(commits []*pt.ParacrossCommitAction) error {
 	for _, cmt := range commits {
 		data := types.Encode(cmt.Status)
@@ -509,6 +531,9 @@ func isCommitDone(nodes, mostSame int) bool {
 }
 
 func (b *blsClient) getBlsPubKey(addr string) (*g2pubs.PublicKey, error) {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
+
 	//先从缓存中获取
 	if v, ok := b.peersBlsPubKey[addr]; ok {
 		return v, nil

plugin/consensus/para/parablssign_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 para
+
+import (
+	"testing"
+
+	pt "github.com/33cn/plugin/plugin/dapp/paracross/types"
+	"github.com/magiconair/properties/assert"
+)
+
+func TestSetAddrsBitMap(t *testing.T) {
+	nodes := []string{"aa", "bb", "cc", "dd"}
+	addrs := []string{}
+	rst, rem := setAddrsBitMap(nodes, addrs)
+	assert.Equal(t, len(rst), 0)
+	assert.Equal(t, len(rem), 0)
+
+	addrs = []string{"aa"}
+	rst, rem = setAddrsBitMap(nodes, addrs)
+	assert.Equal(t, rst, []byte{0x1})
+	assert.Equal(t, len(rem), 0)
+
+	addrs = []string{"aa", "cc"}
+	rst, rem = setAddrsBitMap(nodes, addrs)
+	assert.Equal(t, rst, []byte{0x5})
+	assert.Equal(t, len(rem), 0)
+
+	addrs = []string{"aa", "cc", "dd"}
+	rst, rem = setAddrsBitMap(nodes, addrs)
+	assert.Equal(t, rst, []byte{0xd})
+	assert.Equal(t, len(rem), 0)
+}
+
+func TestIntegrateCommits(t *testing.T) {
+	pool := make(map[int64]*pt.ParaBlsSignSumDetails)
+	var commits []*pt.ParacrossCommitAction
+	cmt1 := &pt.ParacrossCommitAction{
+		Status: &pt.ParacrossNodeStatus{Height: 0},
+		Bls:    &pt.ParacrossCommitBlsInfo{Addrs: []string{"aa"}, Sign: []byte{}},
+	}
+	cmt2 := &pt.ParacrossCommitAction{
+		Status: &pt.ParacrossNodeStatus{Height: 0},
+		Bls:    &pt.ParacrossCommitBlsInfo{Addrs: []string{"bb"}, Sign: []byte{}},
+	}
+	commits = []*pt.ParacrossCommitAction{cmt1, cmt1, cmt1, cmt2, cmt1}
+	integrateCommits(pool, commits)
+	assert.Equal(t, len(pool[0].Addrs), 2)
+	assert.Equal(t, len(pool[0].Msgs), 2)
+	assert.Equal(t, len(pool[0].Signs), 2)
+	assert.Equal(t, pool[0].Addrs[0], "aa")
+	assert.Equal(t, pool[0].Addrs[1], "bb")
+
+}
+
+func TestSecpPrikey2BlsPub(t *testing.T) {
+	key := ""
+	ret, _ := secpPrikey2BlsPub(key)
+	assert.Equal(t, "", ret)
+
+	key = "0xcacb1f5d51700aea07fca2246ab43b0917d70405c65edea9b5063d72eb5c6b71"
+	q := "0x87c58bb6cce41842462a0030335bb95948dcfba77e47e2d8ee893c0b2c34ac20d08c9e98a883ef2a6492d0ad808ace9a1730e8bae5d3b0861aaf743449df5de510073e2991c7274cab47f327e48d7eacf300e4b24174dae2e8603d1904b8a015"
+	ret, _ = secpPrikey2BlsPub(key)
+	assert.Equal(t, q, ret)
+}

plugin/consensus/para/paracommitmsg.go

@@ -328,7 +328,7 @@ func (client *commitMsgClient) checkConsensusStop(checks *commitCheckParams) {
 	if client.sendingHeight > consensHeight {
 		checks.consensStopTimes++
 		if checks.consensStopTimes > client.waitConsensStopTimes {
-			plog.Debug("para commitMsg-checkConsensusStop", "times", checks.consensStopTimes)
+			plog.Debug("para checkConsensusStop", "times", checks.consensStopTimes, "consens", consensHeight, "send", client.sendingHeight)
 			checks.consensStopTimes = 0
 			client.resetSendEnv()
 		}
@@ -961,24 +961,33 @@ func (client *commitMsgClient) onWalletAccount(acc *types.Account) {
 
 }
 
-func (client *commitMsgClient) fetchPriKey() error {
-	req := &types.ReqString{Data: client.authAccount}
-
-	resp, err := client.paraClient.GetAPI().ExecWalletFunc("wallet", "DumpPrivkey", req)
-	str := resp.(*types.ReplyString).Data
-	pk, err := common.FromHex(str)
+func getSecpPriKey(key string) (crypto.PrivKey, error) {
+	pk, err := common.FromHex(key)
 	if err != nil && pk == nil {
-		return err
+		return nil, errors.Wrapf(err, "fromhex=%s", key)
 	}
 
 	secp, err := crypto.New(types.GetSignName("", types.SECP256K1))
 	if err != nil {
-		return err
+		return nil, errors.Wrapf(err, "crypto=%s", key)
 	}
 
 	priKey, err := secp.PrivKeyFromBytes(pk)
 	if err != nil {
-		plog.Error("para commit msg get priKey", "err", err.Error())
+		return nil, errors.Wrapf(err, "fromBytes=%s", key)
+	}
+
+	return priKey, nil
+}
+
+func (client *commitMsgClient) fetchPriKey() error {
+	req := &types.ReqString{Data: client.authAccount}
+
+	resp, err := client.paraClient.GetAPI().ExecWalletFunc("wallet", "DumpPrivkey", req)
+	str := resp.(*types.ReplyString).Data
+	priKey, err := getSecpPriKey(str)
+	if err != nil {
+		plog.Error("para commit msg get priKey", "err", err)
 		return err
 	}
 

plugin/consensus/para/paraquery.go

@@ -5,6 +5,7 @@
 package para
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/33cn/chain33/common"
@@ -72,6 +73,30 @@ func (client *client) Query_CommitTxInfo(req *types.ReqNil) (types.Message, erro
 	return rt, nil
 }
 
+func (client *client) Query_BlsPubKey(req *types.ReqString) (types.Message, error) {
+	if client == nil || req == nil {
+		return nil, fmt.Errorf("%s", "client not bind message queue.")
+	}
+
+	var pub pt.BlsPubKey
+	if len(req.Data) > 0 {
+		p, err := secpPrikey2BlsPub(req.Data)
+		if err != nil {
+			return nil, err
+		}
+		pub.Key = p
+		return &pub, nil
+	}
+	//缺省获取钱包的
+	if nil != client.blsSignCli.blsPubKey {
+		t := client.blsSignCli.blsPubKey.Serialize()
+		pub.Key = common.ToHex(t[:])
+		return &pub, nil
+	}
+
+	return nil, errors.New("no bls prikey init")
+}
+
 // Query_CreateNewAccount 通知para共识模块钱包创建了一个新的账户
 func (client *client) Query_CreateNewAccount(acc *types.Account) (types.Message, error) {
 	if acc == nil {

plugin/dapp/paracross/commands/paracross.go

@@ -44,8 +44,7 @@ func ParcCmd() *cobra.Command {
 		GetBlockInfoCmd(),
 		GetLocalBlockInfoCmd(),
 		GetConsensDoneInfoCmd(),
-		LeaderCmd(),
-		CmtTxInfoCmd(),
+		blsCmd(),
 	)
 	return cmd
 }
@@ -315,7 +314,7 @@ func createCrossAssetTransfer(cmd *cobra.Command, args []string) {
 
 func superNodeCmd() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "super_node",
+		Use:   "node",
 		Short: "super node manage cmd",
 	}
 	cmd.AddCommand(nodeJoinCmd())
@@ -953,11 +952,22 @@ func isSync(cmd *cobra.Command, args []string) {
 	ctx.Run()
 }
 
-// IsSyncCmd query parachain is sync
-func LeaderCmd() *cobra.Command {
+func blsCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "bls",
+		Short: "bls sign manager cmd",
+	}
+	cmd.AddCommand(leaderCmd())
+	cmd.AddCommand(cmtTxInfoCmd())
+	cmd.AddCommand(blsPubKeyCmd())
+	return cmd
+}
+
+// leaderCmd query parachain is sync
+func leaderCmd() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "leader",
-		Short: "node leader info",
+		Short: "current bls sign leader",
 		Run:   leaderInfo,
 	}
 	return cmd
@@ -970,11 +980,11 @@ func leaderInfo(cmd *cobra.Command, args []string) {
 	ctx.Run()
 }
 
-// IsSyncCmd query parachain is sync
-func CmtTxInfoCmd() *cobra.Command {
+// cmtTxInfoCmd query parachain is sync
+func cmtTxInfoCmd() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "cmtinfo",
-		Short: "commit tx info",
+		Use:   "cmts",
+		Short: "current bls sign commits info",
 		Run:   cmtTxInfo,
 	}
 	return cmd
@@ -987,6 +997,27 @@ func cmtTxInfo(cmd *cobra.Command, args []string) {
 	ctx.Run()
 }
 
+// cmtTxInfoCmd query parachain is sync
+func blsPubKeyCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "pubkey",
+		Short: "get bls pub key by secp256 prikey or current wallet bls pubkey",
+		Run:   blsPubKey,
+	}
+	return cmd
+}
+
+func blsPubKey(cmd *cobra.Command, args []string) {
+	cmd.Flags().StringP("prikey", "p", "", "secp256 private key")
+
+	rpcLaddr, _ := cmd.Flags().GetString("rpc_laddr")
+	prikey, _ := cmd.Flags().GetString("prikey")
+	req := &types.ReqString{Data: prikey}
+	var res pt.BlsPubKey
+	ctx := jsonclient.NewRPCCtx(rpcLaddr, "paracross.GetParaBlsPubKey", req, &res)
+	ctx.Run()
+}
+
 func consusHeight(cmd *cobra.Command, args []string) {
 	rpcLaddr, _ := cmd.Flags().GetString("rpc_laddr")
 	paraName, _ := cmd.Flags().GetString("paraName")

plugin/dapp/paracross/executor/action.go

@@ -412,14 +412,9 @@ func getValidAddrs(nodes map[string]struct{}, addrs []string) []string {
 	return ret
 }
 
-//bls签名共识交易验证 大约耗时100ms
-func (a *action) verifyBlsSign(commit *pt.ParacrossCommitAction) ([]string, error) {
-	_, nodesArry, err := a.getNodesGroup(commit.Status.Title)
-	if err != nil {
-		return nil, errors.Wrapf(err, "getNodegroup")
-	}
-
-	//1.　获取addr对应的bls 公钥
+//bls签名共识交易验证 大约平均耗时30ms (20~40ms)
+func (a *action) verifyBlsSign(nodesArry []string, commit *pt.ParacrossCommitAction) ([]string, error) {
+	//1.　获取addr对应的bls 公钥 单独耗时3ms (3 addrs)
 	signAddrs := getAddrsByBitMap(nodesArry, commit.Bls.AddrsMap)
 	var pubs []string
 	for _, addr := range signAddrs {
@@ -429,6 +424,7 @@ func (a *action) verifyBlsSign(commit *pt.ParacrossCommitAction) ([]string, erro
 		}
 		pubs = append(pubs, pub)
 	}
+	//单独deserial 5ms, g2pubs的公钥结构不好整合到protobuf，就不好压缩到数据库直接读取
 	pubKeys := make([]*g2pubs.PublicKey, 0)
 	for _, p := range pubs {
 		k := [96]byte{}
@@ -444,10 +440,11 @@ func (a *action) verifyBlsSign(commit *pt.ParacrossCommitAction) ([]string, erro
 		pubKeys = append(pubKeys, key)
 
 	}
-	//2.　聚合公钥
+
+	//2.　聚合公钥 单独耗时200us
 	aPub := g2pubs.AggregatePublicKeys(pubKeys)
 
-	//3.　获取聚合的签名
+	//3.　获取聚合的签名, deserial 5ms
 	signkey := [48]byte{}
 	copy(signkey[:], commit.Bls.Sign)
 	sign, err := g2pubs.DeserializeSignature(signkey)
@@ -478,27 +475,27 @@ func (a *action) Commit(commit *pt.ParacrossCommitAction) (*types.Receipt, error
 		}
 	}
 
+	nodesMap, nodesArry, err := a.getNodesGroup(commit.Status.Title)
+	if err != nil {
+		return nil, errors.Wrap(err, "getNodesGroup")
+	}
+
 	//获取commitAddrs, bls sign 包含多个账户的聚合签名
 	commitAddrs := []string{a.fromaddr}
 	if commit.Bls != nil {
-		addrs, err := a.verifyBlsSign(commit)
+		addrs, err := a.verifyBlsSign(nodesArry, commit)
 		if err != nil {
 			return nil, errors.Wrap(err, "verifyBlsSign")
 		}
 		commitAddrs = addrs
 	}
 
-	nodesMap, _, err := a.getNodesGroup(commit.Status.Title)
-	if err != nil {
-		return nil, errors.Wrap(err, "getNodesGroup")
-	}
-
-	correctAddrs := getValidAddrs(nodesMap, commitAddrs)
-	if len(correctAddrs) <= 0 {
+	validAddrs := getValidAddrs(nodesMap, commitAddrs)
+	if len(validAddrs) <= 0 {
 		return nil, errors.Wrapf(err, "getValidAddrs nil commitAddrs=%s", strings.Join(commitAddrs, ","))
 	}
 
-	return a.proCommitMsg(commit.Status, nodesMap, correctAddrs)
+	return a.proCommitMsg(commit.Status, nodesMap, validAddrs)
 }
 
 func (a *action) proCommitMsg(commit *pt.ParacrossNodeStatus, nodes map[string]struct{}, commitAddrs []string) (*types.Receipt, error) {

plugin/dapp/paracross/proto/paracross.proto

@@ -449,11 +449,15 @@ message ParaP2PSubMsg {
 
 }
 
+//bls sign leader info
 message ElectionStatus {
     bool   isLeader = 1;
     LeaderSyncInfo leader = 2;
 }
 
+message BlsPubKey{
+    string key = 1;
+}
 
 service paracross {
     rpc IsSync(ReqNil) returns (IsCaughtUp) {}

plugin/dapp/paracross/rpc/rpc.go

@@ -53,7 +53,7 @@ func (c *Jrpc) GetParaLocalBlockInfo(in *types.ReqInt, result *interface{}) erro
 	return nil
 }
 
-// GetParaLocalBlockInfo query para chain the download layer's local height
+// GetParaNodeLeaderInfo query para bls sign leader info
 func (c *channelClient) GetParaNodeLeaderInfo(ctx context.Context, in *types.ReqNil) (*pt.ElectionStatus, error) {
 	data, err := c.QueryConsensusFunc("para", "LeaderInfo", in)
 	if err != nil {
@@ -62,7 +62,7 @@ func (c *channelClient) GetParaNodeLeaderInfo(ctx context.Context, in *types.Req
 	return data.(*pt.ElectionStatus), nil
 }
 
-// GetParaLocalBlockInfo query para local height
+// GetParaNodeLeaderInfo query para bls sign leader info
 func (c *Jrpc) GetParaNodeLeaderInfo(in *types.ReqNil, result *interface{}) error {
 	data, err := c.cli.GetParaNodeLeaderInfo(context.Background(), in)
 	if err != nil {
@@ -72,7 +72,7 @@ func (c *Jrpc) GetParaNodeLeaderInfo(in *types.ReqNil, result *interface{}) erro
 	return nil
 }
 
-// GetParaLocalBlockInfo query para chain the download layer's local height
+// GetParaCmtTxInfo query para chain commit tx info to bls sign
 func (c *channelClient) GetParaCmtTxInfo(ctx context.Context, in *types.ReqNil) (*pt.ParaBlsSignSumInfo, error) {
 	data, err := c.QueryConsensusFunc("para", "CommitTxInfo", in)
 	if err != nil {
@@ -81,7 +81,7 @@ func (c *channelClient) GetParaCmtTxInfo(ctx context.Context, in *types.ReqNil)
 	return data.(*pt.ParaBlsSignSumInfo), nil
 }
 
-// GetParaLocalBlockInfo query para local height
+// GetParaCmtTxInfo query para commit tx info
 func (c *Jrpc) GetParaCmtTxInfo(in *types.ReqNil, result *interface{}) error {
 	data, err := c.cli.GetParaCmtTxInfo(context.Background(), in)
 	if err != nil {
@@ -90,3 +90,22 @@ func (c *Jrpc) GetParaCmtTxInfo(in *types.ReqNil, result *interface{}) error {
 	*result = data
 	return nil
 }
+
+// GetParaBlsPubKey query para chain bls pubkey
+func (c *channelClient) GetParaBlsPubKey(ctx context.Context, in *types.ReqString) (*pt.BlsPubKey, error) {
+	data, err := c.QueryConsensusFunc("para", "BlsPubKey", in)
+	if err != nil {
+		return nil, err
+	}
+	return data.(*pt.BlsPubKey), nil
+}
+
+// GetParaBlsPubKey query para local height
+func (c *Jrpc) GetParaBlsPubKey(in *types.ReqString, result *interface{}) error {
+	data, err := c.cli.GetParaBlsPubKey(context.Background(), in)
+	if err != nil {
+		return err
+	}
+	*result = data
+	return nil
+}