Separate Spam Broadcast Functionality (#1653)

throughput/spam.go

@@ -135,7 +135,6 @@ func (s *Spammer) Spam(ctx context.Context, sh SpamHelper, terminate bool, symbo
 		return err
 	}
 
-	var fundsL sync.Mutex
 	// distribute funds
 	accounts, funds, factories, err := s.distributeFunds(ctx, cli, parser, feePerTx, sh)
 	if err != nil {
@@ -148,23 +147,47 @@ func (s *Spammer) Spam(ctx context.Context, sh SpamHelper, terminate bool, symbo
 		return err
 	}
 
-	// make sure we can exit gracefully & return funds
-	signals := make(chan os.Signal, 2)
-	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
-
 	cctx, cancel := context.WithCancel(ctx)
 	defer cancel()
+
 	for _, issuer := range issuers {
 		issuer.Start(cctx)
 	}
 
 	// set logging
 	issuers[0].logStats(cctx)
 
-	// Broadcast txs
+	// broadcast transactions
+	s.broadcast(cctx, cancel, sh, accounts, funds, factories, issuers, feePerTx, terminate)
+
+	maxUnits, err = chain.EstimateUnits(parser.Rules(time.Now().UnixMilli()), actions, factory)
+	if err != nil {
+		return err
+	}
+	return s.returnFunds(ctx, cli, parser, maxUnits, sh, accounts, factories, funds, symbol)
+}
+
+func (s Spammer) broadcast(
+	ctx context.Context,
+	cancel context.CancelFunc,
+	sh SpamHelper,
+	accounts []*auth.PrivateKey,
+
+	funds map[codec.Address]uint64,
+	factories []chain.AuthFactory,
+	issuers []*issuer,
+
+	feePerTx uint64,
+	terminate bool,
+) {
+	// make sure we can exit gracefully & return funds
+	signals := make(chan os.Signal, 2)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+
 	var (
 		// Do not call this function concurrently (math.Rand is not safe for concurrent use)
-		z = rand.NewZipf(s.zipfSeed, s.sZipf, s.vZipf, uint64(s.numAccounts)-1)
+		z      = rand.NewZipf(s.zipfSeed, s.sZipf, s.vZipf, uint64(s.numAccounts)-1)
+		fundsL = sync.Mutex{}
 
 		it                      = time.NewTimer(0)
 		currentTarget           = min(s.txsPerSecond, s.minTxsPerSecond)
@@ -226,7 +249,7 @@ func (s *Spammer) Spam(ctx context.Context, sh SpamHelper, terminate bool, symbo
 
 					// Send transaction
 					actions := sh.GetTransfer(recipient, amountToTransfer, uniqueBytes())
-					return issuer.Send(cctx, actions, factory, feePerTx)
+					return issuer.Send(ctx, actions, factory, feePerTx)
 				})
 			}
 
@@ -256,7 +279,7 @@ func (s *Spammer) Spam(ctx context.Context, sh SpamHelper, terminate bool, symbo
 				utils.Outf("{{cyan}}increasing target tps:{{/}} %d\n", currentTarget)
 				consecutiveUnderBacklog = 0
 			}
-		case <-cctx.Done():
+		case <-ctx.Done():
 			stop = true
 			utils.Outf("{{yellow}}context canceled{{/}}\n")
 		case <-signals:
@@ -265,14 +288,10 @@ func (s *Spammer) Spam(ctx context.Context, sh SpamHelper, terminate bool, symbo
 			cancel()
 		}
 	}
+
 	// Wait for all issuers to finish
 	utils.Outf("{{yellow}}waiting for issuers to return{{/}}\n")
 	issuerWg.Wait()
-	maxUnits, err = chain.EstimateUnits(parser.Rules(time.Now().UnixMilli()), actions, factory)
-	if err != nil {
-		return err
-	}
-	return s.returnFunds(ctx, cli, parser, maxUnits, sh, accounts, factories, funds, symbol)
 }
 
 func (s *Spammer) logZipf(zipfSeed *rand.Rand) {