test: add the loadtime report tool (#9351) (#9364)

This pull request adds the report tool and modifies the loadtime libraries to better support its use.

(cherry picked from commit 8655080a0f)

Co-authored-by: William Banfield <4561443+williambanfield@users.noreply.github.com>

test/loadtime/report/report.go

@@ -0,0 +1,153 @@
+package report
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/tendermint/tendermint/test/loadtime/payload"
+	"github.com/tendermint/tendermint/types"
+	"gonum.org/v1/gonum/stat"
+)
+
+// BlockStore defines the set of methods needed by the report generator from
+// Tendermint's store.Blockstore type. Using an interface allows for tests to
+// more easily simulate the required behavior without having to use the more
+// complex real API.
+type BlockStore interface {
+	Height() int64
+	Base() int64
+	LoadBlock(int64) *types.Block
+}
+
+// Report contains the data calculated from reading the timestamped transactions
+// of each block found in the blockstore.
+type Report struct {
+	Max, Min, Avg, StdDev time.Duration
+
+	// ErrorCount is the number of parsing errors encountered while reading the
+	// transaction data. Parsing errors may occur if a transaction not generated
+	// by the payload package is submitted to the chain.
+	ErrorCount int
+
+	// NegativeCount is the number of negative durations encountered while
+	// reading the transaction data. A negative duration means that
+	// a transaction timestamp was greater than the timestamp of the block it
+	// was included in and likely indicates an issue with the experimental
+	// setup.
+	NegativeCount int
+
+	// All contains all data points gathered from all valid transactions.
+	// The order of the contents of All is not guaranteed to be match the order of transactions
+	// in the chain.
+	All []time.Duration
+}
+
+// GenerateFromBlockStore creates a Report using the data in the provided
+// BlockStore.
+func GenerateFromBlockStore(s BlockStore) (Report, error) {
+	type payloadData struct {
+		l   time.Duration
+		err error
+	}
+	type txData struct {
+		tx []byte
+		bt time.Time
+	}
+
+	// Deserializing to proto can be slow but does not depend on other data
+	// and can therefore be done in parallel.
+	// Deserializing in parallel does mean that the resulting data is
+	// not guaranteed to be delivered in the same order it was given to the
+	// worker pool.
+	const poolSize = 16
+
+	txc := make(chan txData)
+	pdc := make(chan payloadData, poolSize)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(poolSize)
+	for i := 0; i < poolSize; i++ {
+		go func() {
+			defer wg.Done()
+			for b := range txc {
+				p, err := payload.FromBytes(b.tx)
+				if err != nil {
+					pdc <- payloadData{err: err}
+					continue
+				}
+
+				l := b.bt.Sub(p.Time.AsTime())
+				pdc <- payloadData{l: l}
+			}
+		}()
+	}
+	go func() {
+		wg.Wait()
+		close(pdc)
+	}()
+
+	r := Report{
+		Max: 0,
+		Min: math.MaxInt64,
+	}
+	var sum int64
+	go func() {
+		base, height := s.Base(), s.Height()
+		prev := s.LoadBlock(base)
+		for i := base + 1; i < height; i++ {
+			// Data from two adjacent block are used here simultaneously,
+			// blocks of height H and H+1. The transactions of the block of
+			// height H are used with the timestamp from the block of height
+			// H+1. This is done because the timestamp from H+1 is calculated
+			// by using the precommits submitted at height H. The timestamp in
+			// block H+1 represents the time at which block H was committed.
+			//
+			// In the (very unlikely) event that the very last block of the
+			// chain contains payload transactions, those transactions will not
+			// be used in the latency calculations because the last block whose
+			// transactions are used is the block one before the last.
+			cur := s.LoadBlock(i)
+			for _, tx := range prev.Data.Txs {
+				txc <- txData{tx: tx, bt: cur.Time}
+			}
+			prev = cur
+		}
+		close(txc)
+	}()
+	for pd := range pdc {
+		if pd.err != nil {
+			r.ErrorCount++
+			continue
+		}
+		r.All = append(r.All, pd.l)
+		if pd.l > r.Max {
+			r.Max = pd.l
+		}
+		if pd.l < r.Min {
+			r.Min = pd.l
+		}
+		if int64(pd.l) < 0 {
+			r.NegativeCount++
+		}
+		// Using an int64 here makes an assumption about the scale and quantity of the data we are processing.
+		// If all latencies were 2 seconds, we would need around 4 billion records to overflow this.
+		// We are therefore assuming that the data does not exceed these bounds.
+		sum += int64(pd.l)
+	}
+	if len(r.All) == 0 {
+		r.Min = 0
+		return r, nil
+	}
+	r.Avg = time.Duration(sum / int64(len(r.All)))
+	r.StdDev = time.Duration(int64(stat.StdDev(toFloat(r.All), nil)))
+	return r, nil
+}
+
+func toFloat(in []time.Duration) []float64 {
+	r := make([]float64, len(in))
+	for i, v := range in {
+		r[i] = float64(int64(v))
+	}
+	return r
+}

test/loadtime/report/report_test.go

@@ -0,0 +1,114 @@
+package report_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/tendermint/tendermint/test/loadtime/payload"
+	"github.com/tendermint/tendermint/test/loadtime/report"
+	"github.com/tendermint/tendermint/types"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+type mockBlockStore struct {
+	base   int64
+	blocks []*types.Block
+}
+
+func (m *mockBlockStore) Height() int64 {
+	return m.base + int64(len(m.blocks))
+}
+
+func (m *mockBlockStore) Base() int64 {
+	return m.base
+}
+
+func (m *mockBlockStore) LoadBlock(i int64) *types.Block {
+	return m.blocks[i-m.base]
+}
+
+func TestGenerateReport(t *testing.T) {
+	t1 := time.Now()
+	b1, err := payload.NewBytes(&payload.Payload{
+		Time: timestamppb.New(t1.Add(-10 * time.Second)),
+		Size: 1024,
+	})
+	if err != nil {
+		t.Fatalf("generating payload %s", err)
+	}
+	b2, err := payload.NewBytes(&payload.Payload{
+		Time: timestamppb.New(t1.Add(-4 * time.Second)),
+		Size: 1024,
+	})
+	if err != nil {
+		t.Fatalf("generating payload %s", err)
+	}
+	b3, err := payload.NewBytes(&payload.Payload{
+		Time: timestamppb.New(t1.Add(2 * time.Second)),
+		Size: 1024,
+	})
+	t2 := t1.Add(time.Second)
+	if err != nil {
+		t.Fatalf("generating payload %s", err)
+	}
+	s := &mockBlockStore{
+		blocks: []*types.Block{
+			{
+				Data: types.Data{
+					Txs: []types.Tx{b1, b2},
+				},
+			},
+			{
+				// The timestamp from block H+1 is used to calculate the
+				// latency for the transactions in block H.
+				Header: types.Header{
+					Time: t1,
+				},
+				Data: types.Data{
+					Txs: []types.Tx{[]byte("error")},
+				},
+			},
+			{
+				Data: types.Data{
+					Txs: []types.Tx{b3, b3},
+				},
+			},
+			{
+				Header: types.Header{
+					Time: t2,
+				},
+				Data: types.Data{
+					Txs: []types.Tx{},
+				},
+			},
+		},
+	}
+	r, err := report.GenerateFromBlockStore(s)
+	if err != nil {
+		t.Fatalf("generating report %s", err)
+	}
+	if len(r.All) != 4 {
+		t.Fatalf("report contained different number of data points from expected. Expected %d but contained %d", 4, len(r.All))
+	}
+	if r.ErrorCount != 1 {
+		t.Fatalf("ErrorCount did not match expected. Expected %d but contained %d", 1, r.ErrorCount)
+	}
+	if r.NegativeCount != 2 {
+		t.Fatalf("NegativeCount did not match expected. Expected %d but contained %d", 2, r.NegativeCount)
+	}
+	if r.Avg != 3*time.Second {
+		t.Fatalf("Avg did not match expected. Expected %s but contained %s", 3*time.Second, r.Avg)
+	}
+	if r.Min != -time.Second {
+		t.Fatalf("Min did not match expected. Expected %s but contained %s", time.Second, r.Min)
+	}
+	if r.Max != 10*time.Second {
+		t.Fatalf("Max did not match expected. Expected %s but contained %s", 10*time.Second, r.Max)
+	}
+	// Verified using online standard deviation calculator:
+	// https://www.calculator.net/standard-deviation-calculator.html?numberinputs=10%2C+4%2C+-1%2C+-1&ctype=s&x=45&y=12
+	expectedStdDev := 5228129047 * time.Nanosecond
+	if r.StdDev != expectedStdDev {
+		t.Fatalf("StdDev did not match expected. Expected %s but contained %s", expectedStdDev, r.StdDev)
+	}
+}