tools/tm-bench: Don't count the first block if its empty (#1948)

* tools/tm-bench: Don't count the first block if its empty

* Try melekes suggestion

* Fix getting the start time so the first block is no longer empty

* Fix changelog entry

* Make code smell better

* tools/tmbench: Fix the end time being used for statistics calculation

Previously we were using the time at which all connections closed in statistics, not
the time after {duration} seconds.

* Use waitgroups for starting up

CHANGELOG.md

@@ -16,6 +16,7 @@ BUG FIXES
     - NOTE: this is only for URI requests. JSONRPC requests and all responses
       will use quoted integers (the proto3 JSON standard).
 - [consensus] Fix halt on shutdown
+- [tm_bench] Fix method of computing start time, and end time
 
 ## 0.22.1
 

tools/tm-bench/Gopkg.lock

@@ -119,7 +119,7 @@
     "prometheus",
     "prometheus/promhttp"
   ]
-  revision = "ae27198cdd90bf12cd134ad79d1366a6cf49f632"
+  revision = "ee1c9d7e23df7f011bdf6f12a5c9e7f0ae10a1fe"
 
 [[projects]]
   branch = "master"
@@ -237,7 +237,7 @@
     "types",
     "version"
   ]
-  revision = "9d81a74429e093f3167875e0145ad957874c77d1"
+  revision = "f5ad8ef8600c33532a16de0879ff6b9745bb394d"
 
 [[projects]]
   branch = "master"
@@ -268,7 +268,7 @@
     "netutil",
     "trace"
   ]
-  revision = "292b43bbf7cb8d35ddf40f8d5100ef3837cced3f"
+  revision = "039a4258aec0ad3c79b905677cceeab13b296a77"
 
 [[projects]]
   name = "golang.org/x/text"
@@ -292,10 +292,9 @@
   version = "v0.3.0"
 
 [[projects]]
-  branch = "master"
   name = "google.golang.org/genproto"
   packages = ["googleapis/rpc/status"]
-  revision = "e92b116572682a5b432ddd840aeaba2a559eeff1"
+  revision = "7fd901a49ba6a7f87732eb344f6e3c5b19d1b200"
 
 [[projects]]
   name = "google.golang.org/grpc"
@@ -330,6 +329,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "bc54a74ffdfc09872726fcf5c72b5df882269dc1cd949ac3fbeac9a554fc25c6"
+  inputs-digest = "5c21a60b80ac7d60f7be693de13f9fadb62226b502431bdb38fb9794a98c5b02"
   solver-name = "gps-cdcl"
   solver-version = 1

tools/tm-bench/Gopkg.toml

@@ -45,6 +45,11 @@
   name = "github.com/tendermint/tendermint"
   branch = "develop"
 
+# this got updated and broke, so locked to an old working commit ...
+[[override]]
+  name = "google.golang.org/genproto"
+  revision = "7fd901a49ba6a7f87732eb344f6e3c5b19d1b200"
+
 [prune]
   go-tests = true
   unused-packages = true

tools/tm-bench/README.md

@@ -51,15 +51,26 @@ with the last command being in a seperate window.
 ## How stats are collected
 
 These stats are derived by having each connection send transactions at the
-specified rate (or as close as it can get) for the specified time. After the
-specified time, it iterates over all of the blocks that were created in that
-time. The average and stddev per second are computed based off of that, by
+specified rate (or as close as it can get) for the specified time.
+After the specified time, it iterates over all of the blocks that were created
+in that time.
+The average and stddev per second are computed based off of that, by
 grouping the data by second.
 
 To send transactions at the specified rate in each connection, we loop
-through the number of transactions. If its too slow, the loop stops at one second.
-If its too fast, we wait until the one second mark ends. The transactions per
-second stat is computed based off of what ends up in the block.
+through the number of transactions.
+If its too slow, the loop stops at one second.
+If its too fast, we wait until the one second mark ends.
+The transactions per second stat is computed based off of what ends up in the
+block.
+
+Note that there will be edge effects on the number of transactions in the first
+and last blocks.
+This is because transactions may start sending midway through when tendermint
+starts building the next block, so it only has half as much time to gather txs
+that tm-bench sends.
+Similarly the end of the duration will likely end mid-way through tendermint
+trying to build the next block.
 
 Each of the connections is handled via two separate goroutines. 
 

tools/tm-bench/main.go

@@ -7,6 +7,7 @@ import (
 	"math"
 	"os"
 	"strings"
+	"sync"
 	"text/tabwriter"
 	"time"
 
@@ -25,13 +26,13 @@ type statistics struct {
 }
 
 func main() {
-	var duration, txsRate, connections, txSize int
+	var durationInt, txsRate, connections, txSize int
 	var verbose bool
 	var outputFormat, broadcastTxMethod string
 
 	flagSet := flag.NewFlagSet("tm-bench", flag.ExitOnError)
 	flagSet.IntVar(&connections, "c", 1, "Connections to keep open per endpoint")
-	flagSet.IntVar(&duration, "T", 10, "Exit after the specified amount of time in seconds")
+	flagSet.IntVar(&durationInt, "T", 10, "Exit after the specified amount of time in seconds")
 	flagSet.IntVar(&txsRate, "r", 1000, "Txs per second to send in a connection")
 	flagSet.IntVar(&txSize, "s", 250, "The size of a transaction in bytes.")
 	flagSet.StringVar(&outputFormat, "output-format", "plain", "Output format: plain or json")
@@ -42,7 +43,7 @@ func main() {
 		fmt.Println(`Tendermint blockchain benchmarking tool.
 
 Usage:
-	tm-bench [-c 1] [-T 10] [-r 1000] [endpoints] [-output-format <plain|json> [-broadcast-tx-method <async|sync|commit>]]
+	tm-bench [-c 1] [-T 10] [-r 1000] [-s 250] [endpoints] [-output-format <plain|json> [-broadcast-tx-method <async|sync|commit>]]
 
 Examples:
 	tm-bench localhost:26657`)
@@ -73,7 +74,7 @@ Examples:
 		}
 		logger = log.NewTMLoggerWithColorFn(log.NewSyncWriter(os.Stdout), colorFn)
 
-		fmt.Printf("Running %ds test @ %s\n", duration, flagSet.Arg(0))
+		fmt.Printf("Running %ds test @ %s\n", durationInt, flagSet.Arg(0))
 	}
 
 	if broadcastTxMethod != "async" &&
@@ -93,10 +94,6 @@ Examples:
 	)
 	logger.Info("Latest block height", "h", initialHeight)
 
-	// record time start
-	timeStart := time.Now()
-	logger.Info("Time started", "t", timeStart)
-
 	transacters := startTransacters(
 		endpoints,
 		connections,
@@ -104,9 +101,17 @@ Examples:
 		txSize,
 		"broadcast_tx_"+broadcastTxMethod,
 	)
-	endTime := time.Duration(duration) * time.Second
 
-	<-time.After(endTime)
+	// Wait until transacters have begun until we get the start time
+	timeStart := time.Now()
+	logger.Info("Time last transacter started", "t", timeStart)
+
+	duration := time.Duration(durationInt) * time.Second
+
+	timeEnd := timeStart.Add(duration)
+	logger.Info("End time for calculation", "t", timeEnd)
+
+	<-time.After(duration)
 	for i, t := range transacters {
 		t.Stop()
 		numCrashes := countCrashes(t.connsBroken)
@@ -115,15 +120,14 @@ Examples:
 		}
 	}
 
-	timeStop := time.Now()
-	logger.Info("Time stopped", "t", timeStop)
+	logger.Debug("Time all transacters stopped", "t", time.Now())
 
 	stats, err := calculateStatistics(
 		client,
 		initialHeight,
 		timeStart,
-		timeStop,
-		duration,
+		timeEnd,
+		durationInt,
 	)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
@@ -202,7 +206,7 @@ func calculateStatistics(
 	}
 
 	// iterates from max height to min height
-	for _, blockMeta := range blockMetas {
+	for i, blockMeta := range blockMetas {
 		// check if block was created after timeStart
 		if blockMeta.Header.Time.Before(timeStart) {
 			break
@@ -219,6 +223,7 @@ func calculateStatistics(
 
 		// increase number of txs for that second
 		numTxsPerSec[sec] += blockMeta.Header.NumTxs
+		logger.Debug(fmt.Sprintf("%d txs in block %d, height %d", blockMeta.Header.NumTxs, i, blockMeta.Header.Height))
 	}
 
 	for _, n := range numBlocksPerSec {
@@ -245,15 +250,21 @@ func startTransacters(
 ) []*transacter {
 	transacters := make([]*transacter, len(endpoints))
 
+	wg := sync.WaitGroup{}
+	wg.Add(len(endpoints))
 	for i, e := range endpoints {
 		t := newTransacter(e, connections, txsRate, txSize, broadcastTxMethod)
 		t.SetLogger(logger)
-		if err := t.Start(); err != nil {
-			fmt.Fprintln(os.Stderr, err)
-			os.Exit(1)
-		}
-		transacters[i] = t
+		go func(i int) {
+			defer wg.Done()
+			if err := t.Start(); err != nil {
+				fmt.Fprintln(os.Stderr, err)
+				os.Exit(1)
+			}
+			transacters[i] = t
+		}(i)
 	}
+	wg.Wait()
 
 	return transacters
 }

tools/tm-bench/transacter.go

@@ -36,7 +36,8 @@ type transacter struct {
 
 	conns       []*websocket.Conn
 	connsBroken []bool
-	wg          sync.WaitGroup
+	startingWg  sync.WaitGroup
+	endingWg    sync.WaitGroup
 	stopped     bool
 
 	logger log.Logger
@@ -75,19 +76,22 @@ func (t *transacter) Start() error {
 		t.conns[i] = c
 	}
 
-	t.wg.Add(2 * t.Connections)
+	t.startingWg.Add(t.Connections)
+	t.endingWg.Add(2 * t.Connections)
 	for i := 0; i < t.Connections; i++ {
 		go t.sendLoop(i)
 		go t.receiveLoop(i)
 	}
 
+	t.startingWg.Wait()
+
 	return nil
 }
 
 // Stop closes the connections.
 func (t *transacter) Stop() {
 	t.stopped = true
-	t.wg.Wait()
+	t.endingWg.Wait()
 	for _, c := range t.conns {
 		c.Close()
 	}
@@ -97,7 +101,7 @@ func (t *transacter) Stop() {
 // `broadcast_tx_async`).
 func (t *transacter) receiveLoop(connIndex int) {
 	c := t.conns[connIndex]
-	defer t.wg.Done()
+	defer t.endingWg.Done()
 	for {
 		_, _, err := c.ReadMessage()
 		if err != nil {
@@ -118,6 +122,13 @@ func (t *transacter) receiveLoop(connIndex int) {
 
 // sendLoop generates transactions at a given rate.
 func (t *transacter) sendLoop(connIndex int) {
+	started := false
+	// Close the starting waitgroup, in the event that this fails to start
+	defer func() {
+		if !started {
+			t.startingWg.Done()
+		}
+	}()
 	c := t.conns[connIndex]
 
 	c.SetPingHandler(func(message string) error {
@@ -139,7 +150,7 @@ func (t *transacter) sendLoop(connIndex int) {
 	defer func() {
 		pingsTicker.Stop()
 		txsTicker.Stop()
-		t.wg.Done()
+		t.endingWg.Done()
 	}()
 
 	// hash of the host name is a part of each tx
@@ -156,6 +167,10 @@ func (t *transacter) sendLoop(connIndex int) {
 			startTime := time.Now()
 			endTime := startTime.Add(time.Second)
 			numTxSent := t.Rate
+			if !started {
+				t.startingWg.Done()
+				started = true
+			}
 
 			for i := 0; i < t.Rate; i++ {
 				// each transaction embeds connection index, tx number and hash of the hostname