Add a workload bench tool

sysbench/lua-tests/db/parallel_prepare.lua

@@ -36,15 +36,15 @@ function create_parallel_insert(table_id)
    local c_val
    local pad_val
 
-   for j = 1,oltp_table_size do
+   for j = thread_id*oltp_table_size, (thread_id+1)*oltp_table_size do
 
    c_val = sb_rand_str([[###########-###########-###########-###########-###########-###########-###########-###########-###########-###########]])
    pad_val = sb_rand_str([[###########-###########-###########-###########-###########]])
 
    if (oltp_auto_inc) then
-	 db_bulk_insert_next("(" .. sb_rand(1, oltp_table_size) .. ", '".. c_val .."', '" .. pad_val .. "')")
+	 db_bulk_insert_next("(" .. sb_rand(thread_id*oltp_table_size,(thread_id+1)*oltp_table_size) .. ", '".. c_val .."', '" .. pad_val .. "')")
       else
-	 db_bulk_insert_next("("..j.."," .. sb_rand(1, oltp_table_size) .. ",'".. c_val .."', '" .. pad_val .. "'  )")
+	 db_bulk_insert_next("("..j.."," .. sb_rand(thread_id*oltp_table_size,(thread_id+1)*oltp_table_size) .. ",'".. c_val .."', '" .. pad_val .. "'  )")
       end
    end
 
@@ -59,7 +59,8 @@ end
 function event(thread_id)
    local index_name
    local i
-
+   local tb
+
    log_info("Thread prepare"..thread_id)
 
    if (oltp_secondary) then
@@ -68,7 +69,11 @@ function event(thread_id)
      index_name = "PRIMARY KEY"
    end
 
-   for i=thread_id+1, oltp_tables_count, num_threads  do
-      create_parallel_insert(i)
-   end   
+ --  for i=thread_id+1, oltp_tables_count, num_threads  do
+ --     create_parallel_insert(i)
+ --  end   
+
+   tb = (thread_id%oltp_tables_count)+1
+   create_parallel_insert(tb)
+
 end

workload/db.go

@@ -0,0 +1,77 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"database/sql"
+	"math/rand"
+
+	_ "github.com/go-sql-driver/mysql"
+	log "github.com/sirupsen/logrus"
+)
+
+type DB struct {
+	db         *sql.DB
+	tables     []*Table
+	generators []Generator
+}
+
+func NewDB(dbURL string, dataSize, numTables int) (*DB, error) {
+	db, err := sql.Open("mysql", dbURL)
+	if err != nil {
+		return nil, err
+	}
+
+	db.SetMaxOpenConns(1024)
+	db.SetMaxIdleConns(1024)
+
+	tables := make([]*Table, numTables)
+	generators := make([]Generator, numTables)
+	for i := 0; i < numTables; i++ {
+		tableID := i + 1
+		table, err := NewTable(db, tableID, dataSize)
+		if err != nil {
+			return nil, err
+		}
+		maxID, err := table.MaxID()
+		if err != nil {
+			return nil, err
+		}
+		tables[i] = table
+		generators[i] = NewUniformGenerator(maxID)
+		log.Infof("New tableID %d maxID %d", tableID, maxID)
+	}
+
+	return &DB{
+		db:         db,
+		tables:     tables,
+		generators: generators,
+	}, nil
+}
+
+func (db *DB) rand(r *rand.Rand) (*Table, Generator) {
+	i := r.Int31n(int32(len(db.tables)))
+	return db.tables[i], db.generators[i]
+}
+
+func (db *DB) Insert(r *rand.Rand) error {
+	t, g := db.rand(r)
+	return t.Insert(r, g.NextID())
+}
+
+func (db *DB) Select(r *rand.Rand) error {
+	t, g := db.rand(r)
+	_, err := t.Select(r, g.RandID(r))
+	return err
+}

workload/generator.go

@@ -0,0 +1,41 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"math/rand"
+	"sync/atomic"
+)
+
+type Generator interface {
+	NextID() uint64
+	RandID(*rand.Rand) uint64
+}
+
+type UniformGenerator struct {
+	maxID uint64
+}
+
+func NewUniformGenerator(maxID uint64) *UniformGenerator {
+	return &UniformGenerator{maxID: maxID}
+}
+
+func (g *UniformGenerator) NextID() uint64 {
+	return atomic.AddUint64(&g.maxID, 1)
+}
+
+func (g *UniformGenerator) RandID(r *rand.Rand) uint64 {
+	maxID := atomic.LoadUint64(&g.maxID)
+	return uint64(r.Int63n(int64(maxID + 1)))
+}

workload/main.go

@@ -0,0 +1,91 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"flag"
+	"math/rand"
+	"sync"
+
+	"github.com/BurntSushi/toml"
+	log "github.com/sirupsen/logrus"
+)
+
+type Config struct {
+	DB         string   `toml:"db"`
+	Workload   Workload `toml:"workload"`
+	DataSize   int      `toml:"data-size"`
+	NumTables  int      `toml:"num-tables"`
+	NumWorkers int      `toml:"num-workers"`
+}
+
+type Worker struct {
+	db *DB
+	r  *rand.Rand
+	w  *WorkloadRander
+}
+
+func NewWorker(db *DB, w *WorkloadRander) *Worker {
+	return &Worker{
+		db: db,
+		r:  NewSeekedRander(),
+		w:  w,
+	}
+}
+
+func (w *Worker) Run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		switch w.w.RandomOP(w.r) {
+		case InsertOP:
+			w.db.Insert(w.r)
+		case SelectOP:
+			w.db.Select(w.r)
+		}
+	}
+}
+
+func main() {
+	cfgFile := flag.String("c", "", "configuration file")
+	flag.Parse()
+
+	cfg := new(Config)
+	toml.DecodeFile(*cfgFile, cfg)
+	log.Infof("Load configuration from %s: %+v", *cfgFile, cfg)
+
+	db, err := NewDB(cfg.DB, cfg.DataSize, cfg.NumTables)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	ctx := context.Background()
+
+	for i := 0; i < cfg.NumWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			r := NewWorkloadRander(cfg.Workload)
+			w := NewWorker(db, r)
+			w.Run(ctx)
+		}()
+	}
+
+	wg.Wait()
+}

workload/table.go

@@ -0,0 +1,126 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	TableName   = "twtest"
+	TableSchema = `(
+		id INTEGER UNSIGNED NOT NULL,
+        idx INTEGER UNSIGNED NOT NULL,
+		pad CHAR(32) NOT NULL DEFAULT '',
+        data BLOB,
+		PRIMARY KEY(id), INDEX(idx)
+	)`
+)
+
+func CreateTable(db *sql.DB, table, schema string) error {
+	stmt := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` %s", table, schema)
+	_, err := db.Exec(stmt)
+	return err
+}
+
+func NewSeekedRander() *rand.Rand {
+	return rand.New(rand.NewSource(time.Now().UnixNano()))
+}
+
+const (
+	asciiStart = int('a')
+	asciiLimit = int('z')
+)
+
+func RandomAsciiBytes(r *rand.Rand, size int) []byte {
+	data := make([]byte, size)
+	for i := range data {
+		data[i] = byte(r.Intn(asciiLimit-asciiStart) + asciiStart)
+	}
+	return data
+}
+
+type Table struct {
+	db       *sql.DB
+	table    string
+	dataSize int
+}
+
+func NewTable(db *sql.DB, tableID, dataSize int) (*Table, error) {
+	tableName := fmt.Sprintf("%s%d", TableName, tableID)
+	if err := CreateTable(db, tableName, TableSchema); err != nil {
+		return nil, err
+	}
+
+	return &Table{
+		db:       db,
+		table:    tableName,
+		dataSize: dataSize,
+	}, nil
+}
+
+func (t *Table) MaxID() (uint64, error) {
+	// TODO: use MAX(id)
+	stmt := fmt.Sprintf("SELECT id FROM `%s` ORDER BY id DESC LIMIT 1", t.table)
+	rows, err := t.db.Query(stmt)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var id uint64
+	for rows.Next() {
+		if err := rows.Scan(&id); err != nil {
+			return 0, err
+		}
+	}
+	return id, nil
+}
+
+func (t *Table) Insert(r *rand.Rand, id uint64) error {
+	pad := RandomAsciiBytes(r, 32)
+	data := RandomAsciiBytes(r, t.dataSize)
+	stmt := fmt.Sprintf("INSERT INTO `%s` VALUES (?, ?, ?, ?)", t.table)
+	_, err := t.db.Exec(stmt, id, id, pad, data)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"stmt": stmt,
+			"id":   id,
+		}).Error(err)
+		return err
+	}
+	return nil
+}
+
+func (t *Table) Select(r *rand.Rand, id uint64) (bool, error) {
+	stmt := fmt.Sprintf("SELECT * FROM `%s` WHERE id = ?", t.table)
+	rows, err := t.db.Query(stmt, id)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"stmt": stmt,
+			"id":   id,
+		}).Error(err)
+		return false, err
+	}
+	defer rows.Close()
+	rowsFound := 0
+	for rows.Next() {
+		rowsFound++
+	}
+	return rowsFound != 0, nil
+}