Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(component/bench): report wrong latency for tpcc workload #1577

Merged
merged 2 commits into from
Oct 12, 2021
Merged

fix(component/bench): report wrong latency for tpcc workload #1577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a75c563
fix(component/bench): update go files to match go-tpc module
lobshunter Oct 12, 2021
77d781a
no more panic
lobshunter Oct 12, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 128 additions & 0 deletions components/bench/ch_benchmark.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
package main

import (
"context"
"fmt"
"os"
"runtime"
"strings"
"sync"

"github.com/spf13/cobra"

"github.com/pingcap/go-tpc/ch"
"github.com/pingcap/go-tpc/pkg/workload"
"github.com/pingcap/go-tpc/tpcc"
)

var chConfig ch.Config

func registerCHBenchmark(root *cobra.Command) {
cmd := &cobra.Command{
Use: "ch",
}
cmd.PersistentFlags().IntVar(&tpccConfig.Parts, "parts", 1, "Number to partition warehouses")
cmd.PersistentFlags().IntVar(&tpccConfig.Warehouses, "warehouses", 10, "Number of warehouses")
cmd.PersistentFlags().BoolVar(&tpccConfig.CheckAll, "check-all", false, "Run all consistency checks")
cmd.PersistentFlags().StringVar(&chConfig.RawQueries,
"queries",
"q1,q2,q3,q4,q5,q6,q7,q8,q9,q10,q11,q12,q13,q14,q15,q16,q17,q18,q19,q20,q21,q22",
"All queries")

var cmdPrepare = &cobra.Command{
Use: "prepare",
Short: "Prepare data for the workload",
Run: func(cmd *cobra.Command, args []string) {
executeCH("prepare")
},
}
cmdPrepare.PersistentFlags().BoolVar(&chConfig.CreateTiFlashReplica,
"tiflash",
false,
"Create tiflash replica")

cmdPrepare.PersistentFlags().BoolVar(&chConfig.AnalyzeTable.Enable,
"analyze",
false,
"After data loaded, analyze table to collect column statistics")
// https://pingcap.com/docs/stable/reference/performance/statistics/#control-analyze-concurrency
cmdPrepare.PersistentFlags().IntVar(&chConfig.AnalyzeTable.BuildStatsConcurrency,
"tidb_build_stats_concurrency",
4,
"tidb_build_stats_concurrency param for analyze jobs")
cmdPrepare.PersistentFlags().IntVar(&chConfig.AnalyzeTable.DistsqlScanConcurrency,
"tidb_distsql_scan_concurrency",
15,
"tidb_distsql_scan_concurrency param for analyze jobs")
cmdPrepare.PersistentFlags().IntVar(&chConfig.AnalyzeTable.IndexSerialScanConcurrency,
"tidb_index_serial_scan_concurrency",
1,
"tidb_index_serial_scan_concurrency param for analyze jobs")

var cmdRun = &cobra.Command{
Use: "run",
Short: "Run workload",
Run: func(cmd *cobra.Command, _ []string) {
executeCH("run")
},
}
cmd.AddCommand(cmdRun, cmdPrepare)
root.AddCommand(cmd)
}

func executeCH(action string) {
runtime.GOMAXPROCS(maxProcs)

if err := openDB(); err != nil {
fmt.Println(err)
fmt.Println("Cannot open database, pleae check it (ip/port/username/password)")
os.Exit(1)
}
defer closeDB()

tpccConfig.DBName = dbName
tpccConfig.Threads = threads
tpccConfig.Isolation = isolationLevel
chConfig.DBName = dbName
chConfig.QueryNames = strings.Split(chConfig.RawQueries, ",")

var (
tp, ap workload.Workloader
err error
)
tp, err = tpcc.NewWorkloader(globalDB, &tpccConfig)
if err != nil {
fmt.Printf("Failed to init tp work loader: %v\n", err)
os.Exit(1)
}
ap = ch.NewWorkloader(globalDB, &chConfig)
if err != nil {
fmt.Printf("Failed to init tp work loader: %v\n", err)
os.Exit(1)
}
timeoutCtx, cancel := context.WithTimeout(globalCtx, totalTime)
defer cancel()

if action == "prepare" {
executeWorkload(timeoutCtx, ap, 1, "prepare")
return
}

type workLoaderSetting struct {
workLoader workload.Workloader
threads int
}
var doneWg sync.WaitGroup
for _, workLoader := range []workLoaderSetting{{workLoader: tp, threads: threads}, {workLoader: ap, threads: acThreads}} {
doneWg.Add(1)
go func(workLoader workload.Workloader, threads int) {
executeWorkload(timeoutCtx, workLoader, threads, "run")
doneWg.Done()
}(workLoader.workLoader, workLoader.threads)
}
doneWg.Wait()
fmt.Printf("Finished: %d OLTP workers, %d OLAP workers\n", threads, acThreads)
for _, workLoader := range []workLoaderSetting{{workLoader: tp, threads: threads}, {workLoader: ap, threads: acThreads}} {
workLoader.workLoader.OutputStats(true)
}
}
22 changes: 14 additions & 8 deletions components/bench/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ var (
user string
password string
threads int
acThreads int
driver string
totalTime time.Duration
totalCount int
Expand Down Expand Up @@ -53,11 +54,13 @@ func closeDB() {

func openDB() error {
// TODO: support other drivers
var tmpDB *sql.DB
ds := fmt.Sprintf("%s:%s@tcp(%s:%d)/", user, password, host, port)
dsn := ds + dbName
var err error
globalDB, err = sql.Open(mysqlDriver, dsn)
var (
tmpDB *sql.DB
err error
ds = fmt.Sprintf("%s:%s@tcp(%s:%d)/", user, password, host, port)
)
// allow multiple statements in one query to allow q15 on the TPC-H
globalDB, err = sql.Open(mysqlDriver, fmt.Sprintf("%s%s?multiStatements=true", ds, dbName))
if err != nil {
return err
}
Expand All @@ -74,8 +77,9 @@ func openDB() error {
return err
}
} else {
globalDB.SetMaxIdleConns(threads + 1)
globalDB.SetMaxIdleConns(threads + acThreads + 1)
}

return nil
}

Expand All @@ -92,21 +96,23 @@ func main() {
rootCmd.PersistentFlags().StringVarP(&password, "password", "p", "", "Database password")
rootCmd.PersistentFlags().IntVarP(&port, "port", "P", 4000, "Database port")
rootCmd.PersistentFlags().IntVarP(&threads, "threads", "T", 1, "Thread concurrency")
rootCmd.PersistentFlags().IntVarP(&acThreads, "acThreads", "t", 1, "OLAP client concurrency, only for CH-benCHmark")
rootCmd.PersistentFlags().StringVarP(&driver, "driver", "d", "", "Database driver: mysql")
rootCmd.PersistentFlags().DurationVar(&totalTime, "time", 1<<63-1, "Total execution time")
rootCmd.PersistentFlags().IntVar(&totalCount, "count", 0, "Total execution count, 0 means infinite")
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().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,
rootCmd.PersistentFlags().IntVar(&isolationLevel, "isolation", 0, `Isolation Level 0: Default, 1: ReadUncommitted,
2: ReadCommitted, 3: WriteCommitted, 4: RepeatableRead,
5: Snapshot, 6: Serializable, 7: Linerizable`)

cobra.EnablePrefixMatching = true

registerTpcc(rootCmd)
registerTpch(rootCmd)
registerCHBenchmark(rootCmd)
registerYcsb(rootCmd)

var cancel context.CancelFunc
Expand Down
19 changes: 12 additions & 7 deletions components/bench/misc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package main
import (
"context"
"fmt"
"os"
"sync"
"time"

Expand All @@ -15,6 +16,9 @@ func checkPrepare(ctx context.Context, w workload.Workloader) {
fmt.Println("Skip preparing checking. Please load CSV data into database and check later.")
return
}
if w.Name() == "tpcc" && tpccConfig.NoCheck {
return
}

var wg sync.WaitGroup
wg.Add(threads)
Expand All @@ -34,7 +38,7 @@ func checkPrepare(ctx context.Context, w workload.Workloader) {
wg.Wait()
}

func execute(ctx context.Context, w workload.Workloader, action string, index int) error {
func execute(ctx context.Context, w workload.Workloader, action string, threads, index int) error {
count := totalCount / threads

ctx = w.InitThread(ctx, index)
Expand Down Expand Up @@ -66,7 +70,7 @@ func execute(ctx context.Context, w workload.Workloader, action string, index in

if err != nil {
if !silence {
fmt.Printf("execute %s failed, err %v\n", action, err)
fmt.Printf("[%s] execute %s failed, err %v\n", time.Now().Format("2006-01-02 15:04:05"), action, err)
}
if !ignoreError {
return err
Expand All @@ -77,7 +81,7 @@ func execute(ctx context.Context, w workload.Workloader, action string, index in
return nil
}

func executeWorkload(ctx context.Context, w workload.Workloader, action string) {
func executeWorkload(ctx context.Context, w workload.Workloader, threads int, action string) {
var wg sync.WaitGroup
wg.Add(threads)

Expand All @@ -101,7 +105,11 @@ func executeWorkload(ctx context.Context, w workload.Workloader, action string)
for i := 0; i < threads; i++ {
go func(index int) {
defer wg.Done()
if err := execute(ctx, w, action, index); err != nil {
if err := execute(ctx, w, action, threads, index); err != nil {
if action == "prepare" {
fmt.Printf("a fatal occurred when preparing data: %v\n", err)
os.Exit(1)
}
fmt.Printf("execute %s failed, err %v\n", action, err)
return
}
Expand All @@ -117,7 +125,4 @@ func executeWorkload(ctx context.Context, w workload.Workloader, action string)
outputCancel()

<-ch

fmt.Println("Finished")
w.OutputStats(true)
}
19 changes: 12 additions & 7 deletions components/bench/tpcc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@ import (
_ "net/http/pprof"
"os"
"runtime"
"time"

"github.com/pingcap/go-tpc/pkg/measurement"
"github.com/pingcap/go-tpc/pkg/workload"
"github.com/pingcap/go-tpc/tpcc"
"github.com/spf13/cobra"
Expand All @@ -18,10 +20,7 @@ var tpccConfig tpcc.Config
func executeTpcc(action string) {
if pprofAddr != "" {
go func() {
err := http.ListenAndServe(pprofAddr, http.DefaultServeMux)
if err != nil {
fmt.Printf("failed to ListenAndServe: %s\n", err.Error())
}
http.ListenAndServe(pprofAddr, http.DefaultServeMux)
}()
}
runtime.GOMAXPROCS(maxProcs)
Expand All @@ -31,6 +30,7 @@ func executeTpcc(action string) {
fmt.Println("Cannot open database, pleae check it (ip/port/username/password)")
os.Exit(1)
}
defer closeDB()

tpccConfig.DBName = dbName
tpccConfig.Threads = threads
Expand All @@ -57,9 +57,11 @@ func executeTpcc(action string) {

timeoutCtx, cancel := context.WithTimeout(globalCtx, totalTime)
defer cancel()
defer closeDB()

executeWorkload(timeoutCtx, w, action)
executeWorkload(timeoutCtx, w, threads, action)

fmt.Println("Finished")
w.OutputStats(true)
}

func registerTpcc(root *cobra.Command) {
Expand All @@ -70,19 +72,21 @@ func registerTpcc(root *cobra.Command) {
cmd.PersistentFlags().IntVar(&tpccConfig.Parts, "parts", 1, "Number to partition warehouses")
cmd.PersistentFlags().IntVar(&tpccConfig.Warehouses, "warehouses", 10, "Number of warehouses")
cmd.PersistentFlags().BoolVar(&tpccConfig.CheckAll, "check-all", false, "Run all consistency checks")

var cmdPrepare = &cobra.Command{
Use: "prepare",
Short: "Prepare data for TPCC",
Run: func(cmd *cobra.Command, _ []string) {
executeTpcc("prepare")
},
}
cmdPrepare.PersistentFlags().BoolVar(&tpccConfig.NoCheck, "no-check", false, "TPCC prepare check, default false")
cmdPrepare.PersistentFlags().StringVar(&tpccConfig.OutputType, "output-type", "", "Output file type."+
" If empty, then load data to db. Current only support csv")
cmdPrepare.PersistentFlags().StringVar(&tpccConfig.OutputDir, "output-dir", "", "Output directory for generating file if specified")
cmdPrepare.PersistentFlags().StringVar(&tpccConfig.SpecifiedTables, "tables", "", "Specified tables for "+
"generating file, separated by ','. Valid only if output is set. If this flag is not set, generate all tables by default")
cmdPrepare.PersistentFlags().IntVar(&tpccConfig.PrepareRetryCount, "retry-count", 50, "Retry count when errors occur")
cmdPrepare.PersistentFlags().DurationVar(&tpccConfig.PrepareRetryInterval, "retry-interval", 10*time.Second, "The interval for each retry")

var cmdRun = &cobra.Command{
Use: "run",
Expand All @@ -92,6 +96,7 @@ func registerTpcc(root *cobra.Command) {
},
}
cmdRun.PersistentFlags().BoolVar(&tpccConfig.Wait, "wait", false, "including keying & thinking time described on TPC-C Standard Specification")
cmdRun.PersistentFlags().DurationVar(&tpccConfig.MaxMeasureLatency, "max-measure-latency", measurement.DefaultMaxLatency, "max measure latency in millisecond")

var cmdCleanup = &cobra.Command{
Use: "cleanup",
Expand Down
Loading