Merge pull request #961 from meshplus/feat/mock-test

Feat/for performance test

api/grpc/info.go

@@ -27,10 +27,7 @@ func (cbs *ChainBrokerService) GetTPS(ctx context.Context, req *pb.GetTPSRequest
 		return nil, fmt.Errorf("get tps between %d and %d failed: %w", req.Begin, req.End, err)
 	}
 
-	data := make([]byte, 8)
-	binary.LittleEndian.PutUint64(data, tps)
-
-	return &pb.Response{Data: data}, nil
+	return &pb.Response{Data: []byte(tps)}, nil
 }
 
 func (cbs *ChainBrokerService) GetChainID(ctx context.Context, empty *pb.Empty) (*pb.Response, error) {

cmd/bitxhub/client/chain.go

@@ -21,6 +21,23 @@ func chainCMD() cli.Command {
 				Usage:  "Query BitXHub chain status",
 				Action: getChainStatus,
 			},
+			{
+				Name:  "tps",
+				Usage: "Query BitXHub tps",
+				Flags: []cli.Flag{
+					cli.Uint64Flag{
+						Name:     "begin",
+						Usage:    "Specify begin block number",
+						Required: true,
+					},
+					cli.Uint64Flag{
+						Name:     "end",
+						Usage:    "Specify end block num",
+						Required: true,
+					},
+				},
+				Action: getTps,
+			},
 		},
 	}
 }
@@ -56,3 +73,23 @@ func getChainStatus(ctx *cli.Context) error {
 	return nil
 
 }
+
+func getTps(ctx *cli.Context) error {
+	begin := ctx.Uint64("begin")
+	end := ctx.Uint64("end")
+	url := getURL(ctx, fmt.Sprintf("tps/%d/%d", begin, end))
+
+	data, err := httpGet(ctx, url)
+	if err != nil {
+		return fmt.Errorf("httpGet from url %s failed: %w", url, err)
+	}
+
+	ret, err := parseResponse(data)
+	if err != nil {
+		return fmt.Errorf("wrong response: %w", err)
+	}
+
+	fmt.Println(ret)
+
+	return nil
+}

cmd/bitxhub/client/interchain_manage.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"fmt"
+	"strconv"
 
 	"github.com/fatih/color"
 	"github.com/meshplus/bitxhub-kit/types"
@@ -45,6 +46,18 @@ func interchainMgrCMD() cli.Command {
 				},
 				Action: getIbtpTxHash,
 			},
+			cli.Command{
+				Name:  "status",
+				Usage: "Query tx status by full ibtp id",
+				Flags: []cli.Flag{
+					cli.StringFlag{
+						Name:     "id",
+						Usage:    "Specify full ibtp id",
+						Required: true,
+					},
+				},
+				Action: getIbtpStatus,
+			},
 		},
 	}
 }
@@ -87,3 +100,24 @@ func getIbtpTxHash(ctx *cli.Context) error {
 	}
 	return nil
 }
+
+func getIbtpStatus(ctx *cli.Context) error {
+	id := ctx.String("id")
+
+	receipt, err := invokeBVMContractBySendView(ctx, constant.TransactionMgrContractAddr.String(), "GetStatus", pb.String(id))
+	if err != nil {
+		return fmt.Errorf("invoke BVM contract failed when get ibtp status by id %s: %w", id, err)
+	}
+
+	if receipt.IsSuccess() {
+		status := string(receipt.Ret)
+		res, err := strconv.Atoi(status)
+		if err != nil {
+			return err
+		}
+		fmt.Println("status is:", pb.TransactionStatus(res))
+	} else {
+		color.Red("get interchain counter error: %s\n", string(receipt.Ret))
+	}
+	return nil
+}

go.mod

@@ -20,7 +20,6 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.2
-	github.com/golangci/golangci-lint v1.23.0 // indirect
 	github.com/google/btree v1.0.0
 	github.com/gorilla/mux v1.8.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2
@@ -34,9 +33,9 @@ require (
 	github.com/libp2p/go-libp2p-swarm v0.2.4
 	github.com/looplab/fsm v0.2.0
 	github.com/magiconair/properties v1.8.5
-	github.com/meshplus/bitxhub-core v1.3.1-0.20221027121437-e904eb78f5d5
+	github.com/meshplus/bitxhub-core v1.3.1-0.20230113085923-64d8298ac1d9
 	github.com/meshplus/bitxhub-kit v1.2.1-0.20221104030503-b7107821941e
-	github.com/meshplus/bitxhub-model v1.2.1-0.20230103095329-f8638b97544e
+	github.com/meshplus/bitxhub-model v1.20.2-0.20230113083618-3407ac676767
 	github.com/meshplus/consensus v0.0.0-20211228075008-5f469b198531
 	github.com/meshplus/eth-kit v0.0.0-20221027120404-e69d0c24dbd4
 	github.com/meshplus/go-libp2p-cert v0.0.0-20210125114242-7d9ed2eaaccd

internal/coreapi/api/api.go

@@ -68,7 +68,7 @@ type NetworkAPI interface {
 type ChainAPI interface {
 	Status() string
 	Meta() (*pb.ChainMeta, error)
-	TPS(begin, end uint64) (uint64, error)
+	TPS(begin, end uint64) (string, error)
 }
 
 type FeedAPI interface {

internal/coreapi/chain.go

@@ -28,7 +28,7 @@ func (api *ChainAPI) Meta() (*pb.ChainMeta, error) {
 	return api.bxh.Ledger.GetChainMeta(), nil
 }
 
-func (api *ChainAPI) TPS(begin, end uint64) (uint64, error) {
+func (api *ChainAPI) TPS(begin, end uint64) (string, error) {
 	var (
 		errCount  atomic.Int64
 		total     atomic.Uint64
@@ -39,11 +39,11 @@ func (api *ChainAPI) TPS(begin, end uint64) (uint64, error) {
 	pool := utils.NewGoPool(runtime.GOMAXPROCS(runtime.NumCPU()))
 
 	if int(begin) <= 0 {
-		return 0, fmt.Errorf("begin number should be greater than zero")
+		return "", fmt.Errorf("begin number should be greater than zero")
 	}
 
 	if int(begin) >= int(end) {
-		return 0, fmt.Errorf("begin number should be smaller than end number")
+		return "", fmt.Errorf("begin number should be smaller than end number")
 	}
 
 	// calculate all tx counts
@@ -89,13 +89,14 @@ func (api *ChainAPI) TPS(begin, end uint64) (uint64, error) {
 	pool.Wait()
 
 	if errCount.Load() != 0 {
-		return 0, fmt.Errorf("error during get block TPS")
+		return "", fmt.Errorf("error during get block TPS")
 	}
 
-	elapsed := (endTime - startTime) / int64(time.Second)
+	elapsed := float64(endTime-startTime) / float64(time.Second)
 
 	if elapsed <= 0 {
-		return 0, fmt.Errorf("incorrect block timestamp")
+		return "", fmt.Errorf("incorrect block timestamp")
 	}
-	return total.Load() / uint64(elapsed), nil
+	tps := float64(total.Load()) / elapsed
+	return fmt.Sprintf("total tx count:%d, tps is %f", total.Load(), tps), nil
 }

internal/executor/contracts/interchain.go

@@ -238,11 +238,11 @@ func (x *InterchainManager) checkIBTP(ibtp *pb.IBTP) (*pb.Interchain, bool, *bol
 			isBatch, targetError = x.checkTargetAvailability(srcChainService, dstChainService, ibtp.Type)
 
 			// abandon it because of all service must be ordered
-			// if ordered {
-			if err := checkIndex(interchain.InterchainCounter[dstChainService.getFullServiceId()]+1, ibtp.Index); err != nil {
-				return nil, isBatch, nil, err
+			if !isBatch {
+				if err := checkIndex(interchain.InterchainCounter[dstChainService.getFullServiceId()]+1, ibtp.Index); err != nil {
+					return nil, isBatch, nil, err
+				}
 			}
-			// }
 		} else {
 			if !dstChainService.IsLocal {
 				return nil, isBatch, nil, boltvm.BError(boltvm.InterchainInvalidIBTPNotInCurBXHCode, fmt.Sprintf(string(boltvm.InterchainInvalidIBTPNotInCurBXHMsg), ibtp.ID()))
@@ -254,11 +254,11 @@ func (x *InterchainManager) checkIBTP(ibtp *pb.IBTP) (*pb.Interchain, bool, *bol
 
 			isBatch, targetError = x.checkTargetAvailability(srcChainService, dstChainService, ibtp.Type)
 
-			// if ordered {
-			if err := checkIndex(interchain.InterchainCounter[dstChainService.getFullServiceId()]+1, ibtp.Index); err != nil {
-				return nil, isBatch, nil, err
+			if !isBatch {
+				if err := checkIndex(interchain.InterchainCounter[dstChainService.getFullServiceId()]+1, ibtp.Index); err != nil {
+					return nil, isBatch, nil, err
+				}
 			}
-			// }
 		}
 	} else if ibtp.Category() == pb.IBTP_RESPONSE || isNotification {
 		// Situation which need to check the index
@@ -273,12 +273,10 @@ func (x *InterchainManager) checkIBTP(ibtp *pb.IBTP) (*pb.Interchain, bool, *bol
 				return nil, isBatch, nil, boltvm.BError(boltvm.InterchainInvalidIBTPNotInCurBXHCode, fmt.Sprintf(string(boltvm.InterchainInvalidIBTPNotInCurBXHMsg), ibtp.ID()))
 			}
 		}
-
-		// if dstService == nil || dstService.Ordered {
+		// todo(lrx): adapt receipt batch
 		if err := checkIndex(interchain.ReceiptCounter[dstChainService.getFullServiceId()]+1, ibtp.Index); err != nil {
 			return nil, isBatch, nil, err
 		}
-		// }
 	} else {
 		return nil, isBatch, nil, boltvm.BError(boltvm.InterchainInvalidIBTPIllegalTypeCode, fmt.Sprintf(string(boltvm.InterchainInvalidIBTPIllegalTypeMsg), ibtp.Type))
 	}

internal/executor/contracts/rule_manager_test.go

@@ -598,10 +598,14 @@ BcNwjTDCxyxLNjFKQfMAc6sY6iJs+Ma59WZyC/4uhjE=
 	address, err := privKey.PublicKey().Address()
 	require.Nil(t, err)
 
-	return &repo.Repo{
+	rep := &repo.Repo{
 		Key: &repo.Key{
 			PrivKey: privKey,
 			Address: address.String(),
 		},
+		Config: &repo.Config{},
 	}
+	rep.Config.Executor.Type = "serial"
+
+	return rep
 }

internal/executor/contracts/transaction_manager.go

@@ -438,14 +438,11 @@ func (t *TransactionManager) changeMultiTxStatus(globalID string, txInfo *Transa
 	} else {
 		status := txInfo.ChildTxInfo[txId]
 
-		// if bxh had received child IBTP receipt success, but other child IBTP receipt missing,
+		// if bxh had received child IBTP receipt success/fail/rollback, but other child IBTP receipt missing,
 		// pier need handleMissing all child IBTP because of child IBTP's SrcReceiptCounter does not add 1
-		if status == pb.TransactionStatus_SUCCESS && result == int32(pb.IBTP_RECEIPT_SUCCESS) {
-			status = pb.TransactionStatus_SUCCESS
-		} else {
-			if err := t.setFSM(&status, receipt2EventM[result]); err != nil {
-				return fmt.Errorf("child tx %s with state %v get unexpected receipt %v", txId, status, result)
-			}
+		// so bxh will return err with the ibtp had already reached the final state to pier.
+		if err := t.setFSM(&status, receipt2EventM[result]); err != nil {
+			return fmt.Errorf("child tx %s with state %v get unexpected receipt %v", txId, status, result)
 		}
 
 		txInfo.ChildTxInfo[txId] = status

internal/executor/contracts/transaction_manager_test.go

@@ -2,6 +2,7 @@ package contracts
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/golang/mock/gomock"
@@ -427,7 +428,9 @@ func TestTransactionManager_Report(t *testing.T) {
 	mockStub.EXPECT().SetObject(GlobalTxInfoKey(globalID), txInfo).MaxTimes(1)
 	mockStub.EXPECT().GetObject(GlobalTxInfoKey(globalID), gomock.Any()).SetArg(1, txInfo).Return(true).MaxTimes(1)
 	res = im.Report(id0, int32(pb.IBTP_RECEIPT_SUCCESS))
-	assert.True(t, res.Ok)
+	assert.False(t, res.Ok)
+	const FinalSuccessStatus = "state SUCCESS get unexpected receipt"
+	assert.True(t, strings.Contains(string(res.Result), FinalSuccessStatus))
 
 	txInfo.GlobalState = pb.TransactionStatus_SUCCESS
 	txInfo.ChildTxInfo[id0] = pb.TransactionStatus_BEGIN

internal/executor/executor_test.go

@@ -842,12 +842,16 @@ BcNwjTDCxyxLNjFKQfMAc6sY6iJs+Ma59WZyC/4uhjE=
 	address, err := privKey.PublicKey().Address()
 	require.Nil(t, err)
 
-	return &repo.Repo{
+	rep := &repo.Repo{
 		Key: &repo.Key{
 			PrivKey: privKey,
 			Address: address.String(),
 		},
+		Config: &repo.Config{},
 	}
+	rep.Config.Executor.Type = "serial"
+
+	return rep
 }
 
 func generateMockConfig(t *testing.T) *repo.Config {

internal/ledger/simple_ledger.go

@@ -14,6 +14,8 @@ import (
 
 var _ ledger.StateLedger = (*SimpleLedger)(nil)
 
+const parallel = "parallel"
+
 var (
 	ErrorRollbackToHigherNumber  = fmt.Errorf("rollback to higher blockchain height")
 	ErrorRollbackWithoutJournal  = fmt.Errorf("rollback to blockchain height without journal")
@@ -52,7 +54,8 @@ type SimpleLedger struct {
 	validRevisions     []revision
 	nextRevisionId     int
 
-	changer *stateChanger
+	changer     *stateChanger
+	exeParallel bool
 
 	accessList *ledger.AccessList
 	preimages  map[types.Hash][]byte
@@ -90,6 +93,11 @@ func NewSimpleLedger(repo *repo.Repo, ldb storage.Storage, accountCache *Account
 		}
 	}
 
+	var exeParallel bool
+	if repo.Config.Executor.Type == parallel {
+		exeParallel = true
+	}
+
 	ledger := &SimpleLedger{
 		repo:         repo,
 		logger:       logger,
@@ -103,6 +111,7 @@ func NewSimpleLedger(repo *repo.Repo, ldb storage.Storage, accountCache *Account
 		changer:      newChanger(),
 		accessList:   ledger.NewAccessList(),
 		logs:         NewEvmLogs(),
+		exeParallel:  exeParallel,
 	}
 
 	ledger.changeInstancePool = &sync.Pool{

internal/ledger/simple_ledger_test.go

@@ -1052,7 +1052,9 @@ BcNwjTDCxyxLNjFKQfMAc6sY6iJs+Ma59WZyC/4uhjE=
 			Cert:     repo.Cert{},
 			Txpool:   repo.Txpool{},
 			Order:    repo.Order{},
-			Executor: repo.Executor{},
+			Executor: repo.Executor{
+				Type: "serial",
+			},
 			Ledger: repo.Ledger{
 				Type: "simple",
 			},

internal/ledger/state_accessor.go

@@ -18,8 +18,10 @@ var _ ledger.StateLedger = (*SimpleLedger)(nil)
 
 // GetOrCreateAccount get the account, if not exist, create a new account
 func (l *SimpleLedger) GetOrCreateAccount(addr *types.Address) ledger.IAccount {
-	l.lock.Lock()
-	defer l.lock.Unlock()
+	if l.exeParallel == true {
+		l.lock.Lock()
+		defer l.lock.Unlock()
+	}
 	account := l.GetAccount(addr)
 	if account == nil {
 		account = newAccount(l.ldb, l.accountCache, addr, l.changer)

internal/router/interchain.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/cbergoon/merkletree"
 	"github.com/ethereum/go-ethereum/event"
@@ -100,8 +101,15 @@ func (router *InterchainRouter) PutBlockAndMeta(block *pb.Block, meta *pb.Interc
 	if router.count.Load() == 0 {
 		return
 	}
+	current := time.Now()
 
 	ret := router.classify(block, meta)
+	router.logger.WithFields(logrus.Fields{
+		"height": block.BlockHeader.Number,
+		"count":  len(block.Transactions.Transactions),
+		"elapse": time.Since(current),
+	}).Info("classify block")
+
 	router.piers.Range(func(k, value interface{}) bool {
 		key := k.(string)
 		w := value.(*event.Feed)
@@ -127,6 +135,11 @@ func (router *InterchainRouter) PutBlockAndMeta(block *pb.Block, meta *pb.Interc
 
 		return true
 	})
+	router.logger.WithFields(logrus.Fields{
+		"height": block.BlockHeader.Number,
+		"count":  len(block.Transactions.Transactions),
+		"elapse": time.Since(current),
+	}).Info("end send block to pier")
 }
 
 func (router *InterchainRouter) GetBlockHeader(begin, end uint64, ch chan<- *pb.BlockHeader) error {