From 262c8daa04542d3491074487f09f58c64abd8472 Mon Sep 17 00:00:00 2001 From: Dev Ojha Date: Fri, 13 Jul 2018 01:28:19 -0700 Subject: [PATCH] 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 | 1 + tools/tm-bench/Gopkg.lock | 11 ++++---- tools/tm-bench/Gopkg.toml | 5 ++++ tools/tm-bench/README.md | 23 +++++++++++----- tools/tm-bench/main.go | 51 ++++++++++++++++++++++-------------- tools/tm-bench/transacter.go | 25 ++++++++++++++---- 6 files changed, 79 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fcab4b83..698b8226a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/tools/tm-bench/Gopkg.lock b/tools/tm-bench/Gopkg.lock index 175acb3a6..aa1d819ca 100644 --- a/tools/tm-bench/Gopkg.lock +++ b/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 diff --git a/tools/tm-bench/Gopkg.toml b/tools/tm-bench/Gopkg.toml index 18498cbbb..3b2dfa4ec 100644 --- a/tools/tm-bench/Gopkg.toml +++ b/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 diff --git a/tools/tm-bench/README.md b/tools/tm-bench/README.md index 811141629..000f20f37 100644 --- a/tools/tm-bench/README.md +++ b/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. diff --git a/tools/tm-bench/main.go b/tools/tm-bench/main.go index 4bc67ab38..5f597cecb 100644 --- a/tools/tm-bench/main.go +++ b/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 [-broadcast-tx-method ]] + tm-bench [-c 1] [-T 10] [-r 1000] [-s 250] [endpoints] [-output-format [-broadcast-tx-method ]] 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 } diff --git a/tools/tm-bench/transacter.go b/tools/tm-bench/transacter.go index de408136d..2834727b5 100644 --- a/tools/tm-bench/transacter.go +++ b/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