diff --git a/sysbench/lua-tests/db/parallel_prepare.lua b/sysbench/lua-tests/db/parallel_prepare.lua index 320b268..aa73017 100644 --- a/sysbench/lua-tests/db/parallel_prepare.lua +++ b/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 diff --git a/workload/db.go b/workload/db.go new file mode 100644 index 0000000..0595c0c --- /dev/null +++ b/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 +} diff --git a/workload/generator.go b/workload/generator.go new file mode 100644 index 0000000..1774946 --- /dev/null +++ b/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))) +} diff --git a/workload/main.go b/workload/main.go new file mode 100644 index 0000000..ec2b5d8 --- /dev/null +++ b/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() +} diff --git a/workload/table.go b/workload/table.go new file mode 100644 index 0000000..d083962 --- /dev/null +++ b/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 +} diff --git a/workload/workload.go b/workload/workload.go new file mode 100644 index 0000000..e6858cb --- /dev/null +++ b/workload/workload.go @@ -0,0 +1,56 @@ +// 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" +) + +type OP int + +const ( + NOOP OP = iota + InsertOP + SelectOP +) + +type Workload struct { + InsertRatio float32 `toml:"insert-ratio"` + SelectRatio float32 `toml:"select-ratio"` +} + +type WorkloadRander struct { + ratios []float32 +} + +func NewWorkloadRander(w Workload) *WorkloadRander { + return &WorkloadRander{ + ratios: []float32{ + 0.0, + w.InsertRatio, + w.SelectRatio, + }, + } +} + +func (w *WorkloadRander) RandomOP(r *rand.Rand) OP { + p := r.Float32() + for op, ratio := range w.ratios { + if p < ratio { + return OP(op) + } + p -= ratio + } + return NOOP +} diff --git a/workload/workload.toml b/workload/workload.toml new file mode 100644 index 0000000..77124af --- /dev/null +++ b/workload/workload.toml @@ -0,0 +1,8 @@ +db = "root@tcp(127.0.0.1:4000)/test" +data-size = 1000 +num-tables = 1 +num-workers = 1000 + +[workload] +insert-ratio = 0.2 +select-ratio = 0.8