Fixed: error TPM. Added: show current TPM

cmd/go-tpc/main.go

@@ -32,6 +32,7 @@ var (
 	outputInterval time.Duration
 	isolationLevel int
 	silence        bool
+	summaryReport  bool
 	pprofAddr      string
 
 	globalDB  *sql.DB
@@ -97,6 +98,7 @@ func main() {
 	rootCmd.PersistentFlags().BoolVar(&dropData, "dropdata", false, "Cleanup data before prepare")
 	rootCmd.PersistentFlags().BoolVar(&ignoreError, "ignore-error", false, "Ignore error when running workload")
 	rootCmd.PersistentFlags().BoolVar(&silence, "silence", false, "Don't print error when running workload")
+	rootCmd.PersistentFlags().BoolVar(&summaryReport, "summary", false, "Print summary TPM only, or also print current TPM when running workload")
 	rootCmd.PersistentFlags().DurationVar(&outputInterval, "interval", 10*time.Second, "Output interval time")
 	rootCmd.PersistentFlags().IntVar(&isolationLevel, "isolation", 0, `Isolation Level 0: Default, 1: ReadUncommitted, 
 2: ReadCommitted, 3: WriteCommitted, 4: RepeatableRead, 

cmd/go-tpc/misc.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -61,8 +60,7 @@ func execute(ctx context.Context, w workload.Workloader, action string, index in
 		}
 
 		if err != nil {
-			// For TiDB, we may meet too many conflict errors, so here just ignore it
-			if !silence && !strings.Contains(err.Error(), "conflict") {
+			if !silence {
 				fmt.Printf("execute %s failed, err %v\n", action, err)
 			}
 			if !ignoreError {
@@ -90,7 +88,7 @@ func executeWorkload(ctx context.Context, w workload.Workloader, action string)
 				ch <- struct{}{}
 				return
 			case <-ticker.C:
-				measurement.Output()
+				measurement.Output(summaryReport)
 			}
 		}
 	}()
@@ -120,5 +118,5 @@ func executeWorkload(ctx context.Context, w workload.Workloader, action string)
 	<-ch
 
 	fmt.Println("Finished")
-	measurement.Output()
+	measurement.Output(true)
 }

pkg/measurement/hist.go

@@ -55,6 +55,12 @@ func (h *histogram) Measure(latency time.Duration) {
 	h.bucketCount[i] += 1
 }
 
+func (h *histogram) Empty() bool {
+	h.m.Lock()
+	defer h.m.Unlock()
+	return h.count == 0
+}
+
 func (h *histogram) Summary() string {
 	res := h.getInfo()
 

pkg/measurement/measure.go

@@ -12,54 +12,99 @@ import (
 type measurement struct {
 	sync.RWMutex
 
-	opMeasurement map[string]*histogram
+	outputCounter    int64
+	opCurMeasurement map[string]*histogram
+	opSumMeasurement map[string]*histogram
 }
 
-func (m *measurement) measure(op string, lan time.Duration) {
+func (m *measurement) getHist(op string, err error, current bool) *histogram {
+	opMeasurement := m.opSumMeasurement
+	if current {
+		opMeasurement = m.opCurMeasurement
+	}
+
+	// Create hist of {op} and {op}_ERR at the same time, or else the TPM would be incorrect
+	opPairedKey := fmt.Sprintf("%s_ERR", op)
+	if err != nil {
+		op, opPairedKey = opPairedKey, op
+	}
+
 	m.RLock()
-	opM, ok := m.opMeasurement[op]
+	opM, ok := opMeasurement[op]
 	m.RUnlock()
-
 	if !ok {
 		opM = newHistogram()
+		opPairedM := newHistogram()
 		m.Lock()
-		m.opMeasurement[op] = opM
+		opMeasurement[op] = opM
+		opMeasurement[opPairedKey] = opPairedM
 		m.Unlock()
 	}
+	return opM
+}
 
-	opM.Measure(lan)
+func (m *measurement) measure(op string, err error, lan time.Duration) {
+	m.getHist(op, err, true).Measure(lan)
+	m.getHist(op, err, false).Measure(lan)
 }
 
-func (m *measurement) output() {
+func (m *measurement) takeCurMeasurement() (ret map[string]*histogram) {
 	m.RLock()
 	defer m.RUnlock()
-	keys := make([]string, len(m.opMeasurement))
+	ret, m.opCurMeasurement = m.opCurMeasurement, make(map[string]*histogram, 16)
+	return
+}
+
+func outputMeasurement(opMeasurement map[string]*histogram, prefix string) {
+	keys := make([]string, len(opMeasurement))
 	var i = 0
-	for k := range m.opMeasurement {
+	for k := range opMeasurement {
 		keys[i] = k
 		i += 1
 	}
 	sort.Strings(keys)
 
 	for _, op := range keys {
-		fmt.Printf("%-6s - %s\n", strings.ToUpper(op), m.opMeasurement[op].Summary())
+		hist := opMeasurement[op]
+		if !hist.Empty() {
+			fmt.Printf("%s%-6s - %s\n", prefix, strings.ToUpper(op), hist.Summary())
+		}
+	}
+}
+
+func (m *measurement) output(summaryReport bool) {
+	// Clear current measure data every time
+	var opCurMeasurement = m.takeCurMeasurement()
+
+	if summaryReport {
+		m.RLock()
+		defer m.RUnlock()
+		outputMeasurement(m.opSumMeasurement, "[SUM] ")
+	} else {
+		outputMeasurement(opCurMeasurement, "[CUR] ")
+		m.RLock()
+		defer m.RUnlock()
+		m.outputCounter += 1
+		if m.outputCounter%10 == 0 {
+			outputMeasurement(m.opSumMeasurement, "[SUM] ")
+		}
 	}
 }
 
 func (m *measurement) getOpName() []string {
 	m.RLock()
 	defer m.RUnlock()
 
-	res := make([]string, 0, len(m.opMeasurement))
-	for op := range m.opMeasurement {
+	res := make([]string, 0, len(m.opSumMeasurement))
+	for op := range m.opSumMeasurement {
 		res = append(res, op)
 	}
 	return res
 }
 
 // Output prints the measurement summary.
-func Output() {
-	globalMeasure.output()
+func Output(summaryReport bool) {
+	globalMeasure.output(summaryReport)
 }
 
 // EnableWarmUp sets whether to enable warm-up.
@@ -81,12 +126,16 @@ func Measure(op string, lan time.Duration, err error) {
 	if !IsWarmUpFinished() {
 		return
 	}
+	globalMeasure.measure(op, err, lan)
+}
 
-	if err != nil {
-		op = fmt.Sprintf("%s_ERR", op)
+func newMeasurement() *measurement {
+	return &measurement{
+		sync.RWMutex{},
+		0,
+		make(map[string]*histogram, 16),
+		make(map[string]*histogram, 16),
 	}
-
-	globalMeasure.measure(op, lan)
 }
 
 var (
@@ -95,7 +144,6 @@ var (
 )
 
 func init() {
+	globalMeasure = newMeasurement()
 	warmUp = 0
-	globalMeasure = new(measurement)
-	globalMeasure.opMeasurement = make(map[string]*histogram, 16)
 }