From 7ee9bb4ea2c8613b785ef634d3969aaceed04617 Mon Sep 17 00:00:00 2001
From: Anton Kaliaev <anton.kalyaev@gmail.com>
Date: Fri, 17 Mar 2017 00:56:22 +0400
Subject: [PATCH] more accurate statistics

---
 tm-bench/main.go | 118 +++++++++++++++++++++++++++++++++++------------
 1 file changed, 89 insertions(+), 29 deletions(-)

diff --git a/tm-bench/main.go b/tm-bench/main.go
index d91dfe35c..db73ecab8 100644
--- a/tm-bench/main.go
+++ b/tm-bench/main.go
@@ -5,13 +5,22 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"text/tabwriter"
 	"time"
 
+	metrics "github.com/rcrowley/go-metrics"
+	tmtypes "github.com/tendermint/tendermint/types"
 	"github.com/tendermint/tools/tm-monitor/monitor"
 )
 
 var version = "0.1.0.pre"
 
+type statistics struct {
+	BlockTimeSample    metrics.Histogram
+	TxThroughputSample metrics.Histogram
+	BlockLatency       metrics.Histogram
+}
+
 func main() {
 	var listenAddr string
 	var duration, txsRate int
@@ -41,49 +50,100 @@ Examples:
 
 	fmt.Printf("Running %ds test @ %s\n", duration, flag.Arg(0))
 
-	m := startMonitor(flag.Arg(0))
-
 	endpoints := strings.Split(flag.Arg(0), ",")
-	transacters := make([]*transacter, len(endpoints))
-	for i, e := range endpoints {
-		t := newTransacter(e, txsRate)
-		if err := t.Start(); err != nil {
-			panic(err)
-		}
-		transacters[i] = t
+
+	blockCh := make(chan tmtypes.Header, 100)
+	blockLatencyCh := make(chan float64, 100)
+
+	nodes := startNodes(endpoints, blockCh, blockLatencyCh)
+
+	transacters := startTransacters(endpoints, txsRate)
+
+	stats := &statistics{
+		BlockTimeSample:    metrics.NewHistogram(metrics.NewUniformSample(1000)),
+		TxThroughputSample: metrics.NewHistogram(metrics.NewUniformSample(1000)),
+		BlockLatency:       metrics.NewHistogram(metrics.NewUniformSample(1000)),
 	}
 
-	select {
-	case <-time.After(time.Duration(duration) * time.Second):
-		for _, t := range transacters {
-			t.Stop()
+	lastBlockHeight := -1
+
+	durationTimer := time.After(time.Duration(duration) * time.Second)
+	ticker := time.NewTicker(1 * time.Second)
+	var blocks, txs int
+	for {
+		select {
+		case b := <-blockCh:
+			if lastBlockHeight < b.Height {
+				blocks++
+				txs += b.NumTxs
+				lastBlockHeight = b.Height
+			}
+		case l := <-blockLatencyCh:
+			stats.BlockLatency.Update(int64(l))
+		case <-ticker.C:
+			stats.BlockTimeSample.Update(int64(blocks))
+			stats.TxThroughputSample.Update(int64(txs))
+			blocks = 0
+			txs = 0
+		case <-durationTimer:
+			for _, t := range transacters {
+				t.Stop()
+			}
+
+			printStatistics(stats)
+
+			for _, n := range nodes {
+				n.Stop()
+			}
+			return
 		}
-		collectAndPrintResults(m)
-		m.Stop()
 	}
 }
 
-func startMonitor(endpoints string) *monitor.Monitor {
-	m := monitor.NewMonitor()
+func startNodes(endpoints []string, blockCh chan<- tmtypes.Header, blockLatencyCh chan<- float64) []*monitor.Node {
+	nodes := make([]*monitor.Node, len(endpoints))
 
-	for _, e := range strings.Split(endpoints, ",") {
-		if err := m.Monitor(monitor.NewNode(e)); err != nil {
+	for i, e := range endpoints {
+		n := monitor.NewNode(e)
+		n.SendBlocksTo(blockCh)
+		n.SendBlockLatenciesTo(blockLatencyCh)
+		if err := n.Start(); err != nil {
 			panic(err)
 		}
+		nodes[i] = n
 	}
 
-	if err := m.Start(); err != nil {
-		panic(err)
+	return nodes
+}
+
+func startTransacters(endpoints []string, txsRate int) []*transacter {
+	transacters := make([]*transacter, len(endpoints))
+
+	for i, e := range endpoints {
+		t := newTransacter(e, txsRate)
+		if err := t.Start(); err != nil {
+			panic(err)
+		}
+		transacters[i] = t
 	}
 
-	return m
+	return transacters
 }
 
-func collectAndPrintResults(m *monitor.Monitor) {
-	n := m.Network
-	fmt.Println("===")
-	fmt.Printf("Avg block time: %.3f ms\n", n.AvgBlockTime)
-	fmt.Printf("Avg tx throughput: %.0f per sec\n", n.AvgTxThroughput)
-	fmt.Printf("Avg block latency: %.3f ms\n", n.AvgBlockLatency)
-	fmt.Println("===")
+func printStatistics(stats *statistics) {
+	w := tabwriter.NewWriter(os.Stdout, 0, 0, 5, ' ', 0)
+	fmt.Fprintln(w, "Stats\tAvg\tStdev\tMax\t")
+	fmt.Fprintln(w, fmt.Sprintf("Block latency\t%.2fms\t%.2fms\t%dms\t",
+		stats.BlockLatency.Mean()/1000000.0,
+		stats.BlockLatency.StdDev()/1000000.0,
+		stats.BlockLatency.Max()/1000000))
+	fmt.Fprintln(w, fmt.Sprintf("Blocks/sec\t%.3f\t%.3f\t%d\t",
+		stats.BlockTimeSample.Mean(),
+		stats.BlockTimeSample.StdDev(),
+		stats.BlockTimeSample.Max()))
+	fmt.Fprintln(w, fmt.Sprintf("Txs/sec\t%.0f\t%.0f\t%d\t",
+		stats.TxThroughputSample.Mean(),
+		stats.TxThroughputSample.StdDev(),
+		stats.TxThroughputSample.Max()))
+	w.Flush()
 }