batch broadcastTx test

MetaProtocol-main/gen/proto/go/ring/v1/account_req_res.pb.go

@@ -134,7 +134,6 @@ func (m *Sha3Request) GetData() []byte {
 }
 
 type Sha3Response struct {
-	//
 	Hash                 *github_com_CaduceusMetaverseProtocol_MetaTypes_types.Hash `protobuf:"bytes,1,opt,name=hash,proto3,customtype=github.com/CaduceusMetaverseProtocol/MetaTypes/types.Hash" json:"hash,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}                                                   `json:"-"`
 	XXX_unrecognized     []byte                                                     `json:"-"`

MetaProtocol-main/gen/proto/go/ring/v1/service_grpc.pb.go

@@ -24,7 +24,8 @@ const _ = grpc.SupportPackageIsVersion7
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type RingServiceClient interface {
 	Ping(ctx context.Context, in *RequestPing, opts ...grpc.CallOption) (*ResponsePing, error)
-	//rpc BroadcastTx(RequestBroadcastTx) returns (ResponseBroadcastTx);
+	// rpc BroadcastTx(RequestBroadcastTx) returns (ResponseBroadcastTx);
+	BroadcastTxs(ctx context.Context, in *BroadcastEthTxWithFromRequests, opts ...grpc.CallOption) (*ResponseBroadcastTxs, error)
 	BroadcastTx(ctx context.Context, in *BroadcastEthTxWithFromRequest, opts ...grpc.CallOption) (*ResponseBroadcastTx, error)
 	// web3
 	Sha3(ctx context.Context, in *Sha3Request, opts ...grpc.CallOption) (*Sha3Response, error)
@@ -101,6 +102,15 @@ func (c *ringServiceClient) Ping(ctx context.Context, in *RequestPing, opts ...g
 	return out, nil
 }
 
+func (c *ringServiceClient) BroadcastTxs(ctx context.Context, in *BroadcastEthTxWithFromRequests, opts ...grpc.CallOption) (*ResponseBroadcastTxs, error) {
+	out := new(ResponseBroadcastTxs)
+	err := c.cc.Invoke(ctx, "/ring.v1.RingService/BroadcastTxs", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *ringServiceClient) BroadcastTx(ctx context.Context, in *BroadcastEthTxWithFromRequest, opts ...grpc.CallOption) (*ResponseBroadcastTx, error) {
 	out := new(ResponseBroadcastTx)
 	err := c.cc.Invoke(ctx, "/ring.v1.RingService/BroadcastTx", in, out, opts...)
@@ -538,7 +548,8 @@ func (c *ringServiceClient) Logs(ctx context.Context, in *LogsRequest, opts ...g
 // for forward compatibility
 type RingServiceServer interface {
 	Ping(context.Context, *RequestPing) (*ResponsePing, error)
-	//rpc BroadcastTx(RequestBroadcastTx) returns (ResponseBroadcastTx);
+	// rpc BroadcastTx(RequestBroadcastTx) returns (ResponseBroadcastTx);
+	BroadcastTxs(context.Context, *BroadcastEthTxWithFromRequests) (*ResponseBroadcastTxs, error)
 	BroadcastTx(context.Context, *BroadcastEthTxWithFromRequest) (*ResponseBroadcastTx, error)
 	// web3
 	Sha3(context.Context, *Sha3Request) (*Sha3Response, error)
@@ -606,6 +617,9 @@ type UnimplementedRingServiceServer struct {
 func (UnimplementedRingServiceServer) Ping(context.Context, *RequestPing) (*ResponsePing, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
 }
+func (UnimplementedRingServiceServer) BroadcastTxs(context.Context, *BroadcastEthTxWithFromRequests) (*ResponseBroadcastTxs, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method BroadcastTxs not implemented")
+}
 func (UnimplementedRingServiceServer) BroadcastTx(context.Context, *BroadcastEthTxWithFromRequest) (*ResponseBroadcastTx, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method BroadcastTx not implemented")
 }
@@ -781,6 +795,24 @@ func _RingService_Ping_Handler(srv interface{}, ctx context.Context, dec func(in
 	return interceptor(ctx, in, info, handler)
 }
 
+func _RingService_BroadcastTxs_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(BroadcastEthTxWithFromRequests)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(RingServiceServer).BroadcastTxs(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/ring.v1.RingService/BroadcastTxs",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(RingServiceServer).BroadcastTxs(ctx, req.(*BroadcastEthTxWithFromRequests))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _RingService_BroadcastTx_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(BroadcastEthTxWithFromRequest)
 	if err := dec(in); err != nil {
@@ -1656,6 +1688,10 @@ var RingService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "Ping",
 			Handler:    _RingService_Ping_Handler,
 		},
+		{
+			MethodName: "BroadcastTxs",
+			Handler:    _RingService_BroadcastTxs_Handler,
+		},
 		{
 			MethodName: "BroadcastTx",
 			Handler:    _RingService_BroadcastTx_Handler,

main.go

@@ -9,10 +9,10 @@ import (
 )
 
 var (
-	initAcc, initCmp, batchSign, batchRecover, batchRecoverTx, batchVerify bool
-	txCount, goRoutineCount, cmpAmount, batchCount                         int
-	cpuProfile                                                             string
-	cfg                                                                    *tool.Config
+	initAcc, initCmp, broadcastTxArr                                     bool
+	txCount, goRoutineCount, cmpAmount, batchCount, startCount, endCount int
+	cpuProfile                                                           string
+	cfg                                                                  *tool.Config
 )
 
 func init() {
@@ -20,12 +20,11 @@ func init() {
 	initCmd.PersistentFlags().BoolVar(&initAcc, "initAcc", false, "Start after initializing the account")
 	initCmd.PersistentFlags().BoolVar(&initCmp, "initCmp", false, "Start after initializing the account cmp balance")
 	startCmd.PersistentFlags().StringVar(&cpuProfile, "cpuProfile", "cpuProfile.prof", "Statistics cpu profile")
-	startCmd.PersistentFlags().BoolVar(&batchSign, "batchSign", false, "test grpc interface -> batchSign")
-	startCmd.PersistentFlags().BoolVar(&batchRecover, "batchRecover", false, "test grpc interface -> batchRecover")
-	startCmd.PersistentFlags().BoolVar(&batchRecoverTx, "batchRecoverTx", false, "test grpc interface -> batchRecoverTx")
-	startCmd.PersistentFlags().BoolVar(&batchVerify, "batchVerify", false, "test grpc interface -> batchVerify")
+	startCmd.PersistentFlags().IntVar(&startCount, "startCount", 0, "read excel start count")
+	startCmd.PersistentFlags().IntVar(&endCount, "endCount", 100, "read excel end count")
+	startCmd.PersistentFlags().BoolVar(&broadcastTxArr, "broadcastTxArr", false, "test grpc interface -> broadcastTxArr")
 	startCmd.PersistentFlags().IntVar(&txCount, "txCount", 1000, "send tran count")
-	startCmd.PersistentFlags().IntVar(&goRoutineCount, "goRoutineCount", 100, "send tran goRoutine count")
+	startCmd.PersistentFlags().IntVar(&goRoutineCount, "goRoutineCount", 10, "send tran goRoutine count")
 	startCmd.PersistentFlags().IntVar(&batchCount, "batchCount", 100, "batch send tran count")
 	startCmd.AddCommand(initCmd)
 }

startTest.go

@@ -18,35 +18,17 @@ func init() {
 		return
 	}
 	cfg.StorageAccFileName += ".xlsx"
-	SendTxAccountArr = tool.ReadExcel(cfg.Count, cfg.StorageAccFileName)
 }
 
 func startTest() {
+	SendTxAccountArr = tool.ReadExcelOfStartEnd(startCount, endCount, cfg.StorageAccFileName)
 	log.Infof("Program start initAccCount：%d tranCount", cfg.Count)
 	cfg.GoRoutineCount = goRoutineCount
 	cfg.SignCount = txCount / 100
 	cfg.BatchCount = batchCount
-	arr, fromAddrArr := transaction.SignedTxArr(SendTxAccountArr, cfg)
-	if batchSign {
-		if err := transaction.BatchSignHandler(arr, cfg); err != nil {
-			log.Errorf("Bath Send Tran error: %s", err)
-			return
-		}
-	}
-	if batchRecover {
-		if err := transaction.BatchRecoverHandler(arr, cfg); err != nil {
-			log.Errorf("Bath Send Tran error: %s", err)
-			return
-		}
-	}
-	if batchRecoverTx {
-		if err := transaction.BatchRecoverTxHandler(arr, fromAddrArr, cfg); err != nil {
-			log.Errorf("Bath Send Tran error: %s", err)
-			return
-		}
-	}
-	if batchVerify {
-		if err := transaction.BatchVerifyHandler(arr, cfg); err != nil {
+	arr := transaction.SignedTxArr(SendTxAccountArr, cfg)
+	if broadcastTxArr {
+		if err := transaction.BroadcastTxArr(arr, cfg); err != nil {
 			log.Errorf("Bath Send Tran error: %s", err)
 			return
 		}

tool/Account.go

@@ -141,7 +141,7 @@ func ReadExcelOfStartEnd(start int, end int, fileName string) [][]string {
 		return nil
 	}
 	res := make([][]string, 0)
-	for i := start; i <= end; i++ {
+	for i := start; i < end; i++ {
 		res = append(res, rows[i])
 	}
 	return res

transaction/signedTx.go

@@ -4,12 +4,17 @@ import (
 	"ChainGrpcTest/log"
 	"ChainGrpcTest/tool"
 	"ChainGrpcTest/txcache"
+	"context"
 	"crypto/ecdsa"
+	ring "github.com/CaduceusMetaverseProtocol/MetaProtocol/gen/proto/go/ring/v1"
+	metatypes "github.com/CaduceusMetaverseProtocol/MetaTypes/types"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	"math/big"
-	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -28,6 +33,12 @@ type Transactor struct {
 	receivedAddr common.Address
 }
 
+var (
+	tran           = make(chan *Transactor, 0)
+	tranArr        = make([]*types.Transaction, 0)
+	batchSignCount int32
+)
+
 func newTransactor(cfg TranConfig) (*Transactor, error) {
 	signerKey, err := crypto.HexToECDSA(cfg.PrivateKey)
 	if err != nil {
@@ -50,17 +61,29 @@ func newTransactor(cfg TranConfig) (*Transactor, error) {
 	return &res, nil
 }
 
-var accountsNonceMap sync.Map
-
 // SignedTxArr 获取全部签名数据
-func SignedTxArr(sendTxAccountArr [][]string, cfg *tool.Config) ([]*types.Transaction, []common.Address) {
-	tranArr := make([]*types.Transaction, 0)
-	fromAddrArr := make([]common.Address, 0)
-	var signedTx *types.Transaction
+func SignedTxArr(sendTxAccountArr [][]string, cfg *tool.Config) []*types.Transaction {
+	for i := 0; i < cfg.GoRoutineCount; i++ {
+		go signedTxFunc()
+	}
+	client, err := grpc.Dial(cfg.RpcNode, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Error("grpc dial error:", err)
+		return nil
+	}
+	defer client.Close()
+	serviceClient := ring.NewRingServiceClient(client)
 	for _, rows := range sendTxAccountArr {
-		fromAddrArr = append(fromAddrArr, common.HexToAddress(rows[0]))
 		privateKey := rows[1]
-		nonce := big.NewInt(1)
+		fromAddr := metatypes.HexToAddress(rows[0])
+		nonceReq := &ring.NonceRequest{
+			Address: &fromAddr,
+		}
+		response, err := serviceClient.Nonce(context.Background(), nonceReq, nil)
+		if err != nil {
+			return nil
+		}
+		nonce := new(big.Int).SetUint64(response.Nonce)
 		for signCount := 0; signCount < cfg.SignCount; signCount++ {
 			tranCfg := TranConfig{
 				Amount:         cfg.Amount,
@@ -71,35 +94,44 @@ func SignedTxArr(sendTxAccountArr [][]string, cfg *tool.Config) ([]*types.Transa
 				Nonce:          nonce,
 			}
 			t, err := newTransactor(tranCfg)
-			signedTx, err = t.signedTx()
-			txcache.Add(signedTx.Hash().Hex(), rows[0])
-			nonce = big.NewInt(1).Add(nonce, big.NewInt(1))
-			if err != nil || signedTx == nil {
+			if err != nil {
 				log.Errorf("signed tx error %s ", err)
 				continue
 			}
-			tranArr = append(tranArr, signedTx)
+			tran <- t
+			nonce = big.NewInt(1).Add(nonce, big.NewInt(1))
+		}
+	}
+	for {
+		if len(sendTxAccountArr)*cfg.SignCount == int(batchSignCount) {
+			return tranArr
 		}
 	}
-	return tranArr, fromAddrArr
 }
 
-// signedTx 签名本币转账交易
-func (t *Transactor) signedTx() (*types.Transaction, error) {
-	txData := types.LegacyTx{
-		Nonce:    t.config.Nonce.Uint64(),
-		To:       &t.receivedAddr,
-		Value:    big.NewInt(t.config.Amount),
-		Gas:      300000,
-		GasPrice: big.NewInt(1000000001),
-		Data:     nil,
-	}
-	newtx := types.NewTx(&txData)
-	signedTx, err := types.SignTx(newtx, types.NewEIP155Signer(big.NewInt(t.config.ChainId)), t.signerKey)
-	if err != nil {
-		log.Errorf("Send tx nonce: %d , From: %s , to: %s , error: %s", t.config.Nonce, crypto.PubkeyToAddress(t.signerKey.PublicKey), t.receivedAddr, err.Error())
-		time.Sleep(time.Second)
-		return nil, err
+// signedTxFunc 签名本币转账交易
+func signedTxFunc() (*types.Transaction, error) {
+	for {
+		select {
+		case t := <-tran:
+			txData := types.LegacyTx{
+				Nonce:    t.config.Nonce.Uint64(),
+				To:       &t.receivedAddr,
+				Value:    big.NewInt(t.config.Amount),
+				Gas:      300000,
+				GasPrice: big.NewInt(1000000001),
+				Data:     nil,
+			}
+			newtx := types.NewTx(&txData)
+			signedTx, err := types.SignTx(newtx, types.NewEIP155Signer(big.NewInt(t.config.ChainId)), t.signerKey)
+			if err != nil {
+				log.Errorf("Send tx nonce: %d , From: %s , to: %s , error: %s", t.config.Nonce, crypto.PubkeyToAddress(t.signerKey.PublicKey), t.receivedAddr, err.Error())
+				time.Sleep(time.Second)
+				return nil, err
+			}
+			txcache.Add(signedTx.Hash().Hex(), t.sender.Hex())
+			tranArr = append(tranArr, signedTx)
+			atomic.AddInt32(&batchSignCount, 1)
+		}
 	}
-	return signedTx, nil
 }

txcache/txcache.go

@@ -6,7 +6,7 @@ import (
 )
 
 var (
-	gtxcache = &TxCache{
+	gtxCache = &TxCache{
 		cache: make(map[string]string),
 	}
 )
@@ -16,18 +16,25 @@ type TxCache struct {
 	mux   sync.RWMutex
 }
 
-func Add(txhash string, address string) {
-	gtxcache.mux.Lock()
-	defer gtxcache.mux.Unlock()
-	gtxcache.cache[txhash] = address
+func Add(txHash string, address string) {
+	gtxCache.mux.Lock()
+	defer gtxCache.mux.Unlock()
+	gtxCache.cache[txHash] = address
 }
 
-func CompareAddress(txhash string, newer string) bool {
-	gtxcache.mux.RLock()
-	defer gtxcache.mux.RUnlock()
-	o, exist := gtxcache.cache[txhash]
+func CompareAddress(txHash string, newer string) bool {
+	gtxCache.mux.RLock()
+	defer gtxCache.mux.RUnlock()
+	o, exist := gtxCache.cache[txHash]
 	if !exist {
 		return false
 	}
 	return strings.Compare(newer, o) == 0
 }
+
+func GetFromAddr(txHash string) (string, bool) {
+	gtxCache.mux.RLock()
+	defer gtxCache.mux.RUnlock()
+	o, exist := gtxCache.cache[txHash]
+	return o, exist
+}