@@ -3,22 +3,21 @@ package main
33import (
44 "context"
55 "fmt"
6+ "math"
67 "math/rand"
7- "os"
8- "path/filepath"
98 "strings"
109 "sync"
10+ "sync/atomic"
1111 "time"
1212
13- cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
13+ "golang.org/x/time/rate"
14+
1415 "github.com/cosmos/cosmos-sdk/types"
15- sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
1616 typestx "github.com/cosmos/cosmos-sdk/types/tx"
1717 stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
1818
1919 "crypto/tls"
2020
21- "github.com/k0kubun/pp/v3"
2221 "google.golang.org/grpc"
2322 "google.golang.org/grpc/credentials"
2423)
@@ -27,12 +26,8 @@ type LoadTestClient struct {
2726 LoadTestConfig Config
2827 TestConfig EncodingConfig
2928 TxClients []typestx.ServiceClient
30- TxHashFile * os.File
3129 SignerClient * SignerClient
3230 ChainID string
33- TxHashList []string
34- TxResponseChan chan * string
35- TxHashListMutex * sync.Mutex
3631 GrpcConns []* grpc.ClientConn
3732 StakingQueryClient stakingtypes.QueryClient
3833 // Staking specific variables
@@ -67,22 +62,12 @@ func NewLoadTestClient(config Config) *LoadTestClient {
6762 GrpcConns [i ] = grpcConn
6863 }
6964
70- // setup output files
71- userHomeDir , _ := os .UserHomeDir ()
72- _ = os .Mkdir (filepath .Join (userHomeDir , "outputs" ), os .ModePerm )
73- filename := filepath .Join (userHomeDir , "outputs" , "test_tx_hash" )
74- _ = os .Remove (filename )
75- outputFile , _ := os .OpenFile (filename , os .O_APPEND | os .O_CREATE | os .O_WRONLY , 0o644 )
7665 return & LoadTestClient {
7766 LoadTestConfig : config ,
7867 TestConfig : TestConfig ,
7968 TxClients : TxClients ,
80- TxHashFile : outputFile ,
8169 SignerClient : NewSignerClient (config .NodeURI ),
8270 ChainID : config .ChainID ,
83- TxHashList : []string {},
84- TxResponseChan : make (chan * string ),
85- TxHashListMutex : & sync.Mutex {},
8671 GrpcConns : GrpcConns ,
8772 StakingQueryClient : stakingtypes .NewQueryClient (GrpcConns [0 ]),
8873 DelegationMap : map [string ]map [string ]int {},
@@ -108,194 +93,53 @@ func (c *LoadTestClient) Close() {
10893 }
10994}
11095
111- func (c * LoadTestClient ) AppendTxHash (txHash string ) {
112- c .TxResponseChan <- & txHash
113- }
114-
115- func (c * LoadTestClient ) WriteTxHashToFile () {
116- fmt .Printf ("Writing Tx Hashes to: %s \n " , c .TxHashFile .Name ())
117- file := c .TxHashFile
118- for _ , txHash := range c .TxHashList {
119- txHashLine := fmt .Sprintf ("%s\n " , txHash )
120- if _ , err := file .WriteString (txHashLine ); err != nil {
121- panic (err )
122- }
123- }
124- }
125-
126- func (c * LoadTestClient ) BuildTxs () (workgroups []* sync.WaitGroup , sendersList [][]func ()) {
96+ func (c * LoadTestClient ) BuildTxs (txQueue chan <- []byte , producerId int , wg * sync.WaitGroup , done <- chan struct {}, producedCount * int64 ) {
97+ defer wg .Done ()
12798 config := c .LoadTestConfig
128- numberOfAccounts := config .TxsPerBlock / config .MsgsPerTx * 2 // * 2 because we need two sets of accounts
129- activeAccounts := []int {}
130- inactiveAccounts := []int {}
131-
132- for i := 0 ; i < int (numberOfAccounts ); i ++ {
133- if i % 2 == 0 {
134- activeAccounts = append (activeAccounts , i )
135- } else {
136- inactiveAccounts = append (inactiveAccounts , i )
137- }
138- }
139-
140- valKeys := c .SignerClient .GetValKeys ()
141-
142- for i := 0 ; i < int (config .Rounds ); i ++ {
143- fmt .Printf ("Preparing %d-th round\n " , i )
144-
145- wg := & sync.WaitGroup {}
146- var senders []func ()
147- workgroups = append (workgroups , wg )
148- c .generatedAdminMessageForBlock = false
149- for j , account := range activeAccounts {
150- accountIdentifier := fmt .Sprint (account )
151- accountKeyPath := c .SignerClient .GetTestAccountKeyPath (uint64 (account ))
152- key := c .SignerClient .GetKey (accountIdentifier , "test" , accountKeyPath )
153-
154- msgs , failureExpected , signer , gas , fee := c .generateMessage (config , key , config .MsgsPerTx )
99+ accountIdentifier := fmt .Sprint (producerId )
100+ accountKeyPath := c .SignerClient .GetTestAccountKeyPath (uint64 (producerId ))
101+ key := c .SignerClient .GetKey (accountIdentifier , "test" , accountKeyPath )
102+
103+ for {
104+ select {
105+ case <- done :
106+ fmt .Printf ("Stopping producer %d\n " , producerId )
107+ return
108+ default :
109+ msgs , _ , _ , gas , fee := c .generateMessage (config , key , config .MsgsPerTx )
155110 txBuilder := TestConfig .TxConfig .NewTxBuilder ()
156111 _ = txBuilder .SetMsgs (msgs ... )
157- seqDelta := uint64 (i / 2 )
158- mode := typestx .BroadcastMode_BROADCAST_MODE_SYNC
159- if j == len (activeAccounts )- 1 {
160- mode = typestx .BroadcastMode_BROADCAST_MODE_BLOCK
161- }
162- // Note: There is a potential race condition here with seqnos
163- // in which a later seqno is delievered before an earlier seqno
164- // In practice, we haven't run into this issue so we'll leave this
165- // as is.
166- sender := SendTx (signer , & txBuilder , mode , seqDelta , failureExpected , * c , gas , fee )
167- wg .Add (1 )
168- senders = append (senders , func () {
169- defer wg .Done ()
170- sender ()
171- })
172- }
173-
174- senders = append (senders , c .GenerateOracleSenders (i , config , valKeys , wg )... )
175-
176- sendersList = append (sendersList , senders )
177- inactiveAccounts , activeAccounts = activeAccounts , inactiveAccounts
178- }
179-
180- return workgroups , sendersList
181- }
182-
183- func (c * LoadTestClient ) GenerateOracleSenders (i int , config Config , valKeys []cryptotypes.PrivKey , waitGroup * sync.WaitGroup ) []func () {
184- senders := []func (){}
185- if config .RunOracle && i % 2 == 0 {
186- for _ , valKey := range valKeys {
187- // generate oracle tx
188- msg := generateOracleMessage (valKey )
189- txBuilder := TestConfig .TxConfig .NewTxBuilder ()
190- _ = txBuilder .SetMsgs (msg )
191- seqDelta := uint64 (i / 2 )
192- mode := typestx .BroadcastMode_BROADCAST_MODE_SYNC
193- sender := SendTx (valKey , & txBuilder , mode , seqDelta , false , * c , 30000 , 100000 )
194- waitGroup .Add (1 )
195- senders = append (senders , func () {
196- defer waitGroup .Done ()
197- sender ()
112+ txBuilder .SetGasLimit (gas )
113+ txBuilder .SetFeeAmount ([]types.Coin {
114+ types .NewCoin ("usei" , types .NewInt (fee )),
198115 })
199- }
200- }
201- return senders
202- }
203-
204- func (c * LoadTestClient ) SendTxs (workgroups []* sync.WaitGroup , sendersList [][]func ()) {
205- defer close (c .TxResponseChan )
116+ // Use random seqno to get around txs that might already be seen in mempool
206117
207- lastHeight := getLastHeight (c .LoadTestConfig .BlockchainEndpoint )
208- for i := 0 ; i < int (c .LoadTestConfig .Rounds ); i ++ {
209- newHeight := getLastHeight (c .LoadTestConfig .BlockchainEndpoint )
210- for newHeight == lastHeight {
211- time .Sleep (10 * time .Millisecond )
212- newHeight = getLastHeight (c .LoadTestConfig .BlockchainEndpoint )
213- }
214- fmt .Printf ("Sending %d-th block\n " , i )
215- senders := sendersList [i ]
216- wg := workgroups [i ]
217- for _ , sender := range senders {
218- go sender ()
118+ c .SignerClient .SignTx (c .ChainID , & txBuilder , key , uint64 (rand .Intn (math .MaxInt )))
119+ txBytes , _ := TestConfig .TxConfig .TxEncoder ()(txBuilder .GetTx ())
120+ txQueue <- txBytes
121+ atomic .AddInt64 (producedCount , 1 )
219122 }
220- wg .Wait ()
221- lastHeight = newHeight
222123 }
223124}
224125
225- func (c * LoadTestClient ) GatherTxHashes () {
226- for txHash := range c .TxResponseChan {
227- c .TxHashList = append (c .TxHashList , * txHash )
228- }
229- fmt .Printf ("Transactions Sent=%d\n " , len (c .TxHashList ))
230- }
231-
232- func (c * LoadTestClient ) ValidateTxs () {
233- defer c .Close ()
234- numTxs := len (c .TxHashList )
235- resultChan := make (chan * types.TxResponse , numTxs )
236- var waitGroup sync.WaitGroup
237-
238- if numTxs == 0 {
239- return
240- }
241-
242- for _ , txHash := range c .TxHashList {
243- waitGroup .Add (1 )
244- go func (txHash string ) {
245- defer waitGroup .Done ()
246- resultChan <- c .GetTxResponse (txHash )
247- }(txHash )
248- }
249-
250- go func () {
251- waitGroup .Wait ()
252- close (resultChan )
253- }()
254-
255- fmt .Printf ("Validating %d Transactions... \n " , len (c .TxHashList ))
256- waitGroup .Wait ()
257-
258- notCommittedTxs := 0
259- responseCodeMap := map [int ]int {}
260- responseStringMap := map [string ]int {}
261- for result := range resultChan {
262- // If the result is nil then that means the transaction was not committed
263- if result == nil {
264- notCommittedTxs ++
265- continue
266- }
267- code := result .Code
268- codeString := "ok"
269- if code != 0 {
270- codespace := result .Codespace
271- err := sdkerrors .ABCIError (codespace , code , fmt .Sprintf ("Error code=%d " , code ))
272- codeString = err .Error ()
126+ func (c * LoadTestClient ) SendTxs (txQueue <- chan []byte , done <- chan struct {}, sentCount * int64 , rateLimit int ) {
127+ rateLimiter := rate .NewLimiter (rate .Limit (rateLimit ), rateLimit )
128+ for {
129+
130+ select {
131+ case <- done :
132+ fmt .Printf ("Stopping consumers\n " )
133+ return
134+ case tx , ok := <- txQueue :
135+ if ! ok {
136+ fmt .Printf ("Stopping consumers\n " )
137+ }
138+ if rateLimiter .Allow () {
139+ go SendTx (tx , typestx .BroadcastMode_BROADCAST_MODE_BLOCK , false , * c , sentCount )
140+ }
273141 }
274- responseStringMap [codeString ]++
275- responseCodeMap [int (code )]++
276- }
277-
278- fmt .Printf ("Transactions not committed: %d\n " , notCommittedTxs )
279- pp .Printf ("Response Code Mapping: \n %s \n " , responseStringMap )
280- IncrTxNotCommitted (notCommittedTxs )
281- for reason , count := range responseStringMap {
282- IncrTxProcessCode (reason , count )
283- }
284- }
285-
286- func (c * LoadTestClient ) GetTxResponse (hash string ) * types.TxResponse {
287- grpcRes , err := c .GetTxClient ().GetTx (
288- context .Background (),
289- & typestx.GetTxRequest {
290- Hash : hash ,
291- },
292- )
293- fmt .Printf ("Validated: %s\n " , hash )
294- if err != nil {
295- fmt .Println (err )
296- return nil
297142 }
298- return grpcRes .TxResponse
299143}
300144
301145func (c * LoadTestClient ) GetTxClient () typestx.ServiceClient {
0 commit comments