Sysbench (#206)

### What problem were solved in this pull request?

Issue Number: close #178 

Problem:
sysbench is a powerful concurrency test tool and we should use it to
test our program.

### What is changed and how it works?
I create two sysbench lua scripts and a github workflow.

### Other information

.github/workflows/build.yml

@@ -52,17 +52,3 @@ jobs:
     - name: Build
       shell: bash
       run: sudo bash build.sh init && bash build.sh release --make -j4
-      
-  basic-test:
-    runs-on: ubuntu-latest
-
-    steps:
-      - name: Checkout repository and submodules
-        uses: actions/checkout@v2
-
-      - name: run basic test
-        shell: bash
-        run: |
-          sudo bash build.sh init
-          echo "begin test..."
-          python3 test/case/miniob_test.py --test-cases=basic | tail -1 | grep "basic is success"

.github/workflows/test.yml

+name: test
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+env:
+  # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)
+  BUILD_TYPE: Release
+
+jobs:
+  basic-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository and submodules
+        uses: actions/checkout@v2
+
+      - name: run basic test
+        shell: bash
+        run: |
+          sudo bash build.sh init
+          echo "begin test..."
+          python3 test/case/miniob_test.py --test-cases=basic | tail -1 | grep "basic is success"
+
+  # sysbench cannot work property on this platform.
+  # I found that sysbench would send more request before receiving last response
+  # sysbench-test:
+  #   runs-on: ubuntu-latest
+
+
+  #   steps:
+  #     - name: Checkout repository and submodules
+  #       uses: actions/checkout@v2
+
+  #     - name: install sysbench and mariadb-client
+  #       shell: bash
+  #       run: |
+  #         curl -s https://packagecloud.io/install/repositories/akopytov/sysbench/script.deb.sh -o script.deb.sh
+  #         sudo bash script.deb.sh
+  #         sudo apt -y install sysbench mariadb-client
+
+  #     - name: start server
+  #       shell: bash
+  #       run: |
+  #         sudo bash build.sh init
+  #         bash build.sh -DCONCURRENCY=ON -DWITH_UNIT_TESTS=OFF
+  #         nohup ./build_debug/bin/observer -s /tmp/miniob.sock -f etc/observer.ini -P mysql -t mvcc &
+  #         sleep 10 && echo "wake up"
+  #         mysql --version
+  #         mysql -S /tmp/miniob.sock -e "show tables"
+
+  #     - name: sysbench test
+  #       shell: bash
+  #       run: |
+  #         cd test/sysbench
+  #         sysbench --mysql-socket=/tmp/miniob.sock --threads=10 miniob_insert prepare
+  #         sysbench --mysql-socket=/tmp/miniob.sock --threads=10 miniob_insert run

deps/common/lang/mutex.h

@@ -19,6 +19,7 @@ See the Mulan PSL v2 for more details. */
 #include <pthread.h>
 #include <string.h>
 #include <unordered_map>
+#include <condition_variable>
 #include <map>
 #include <sstream>
 #include <string>

deps/common/os/signal.cpp

@@ -38,6 +38,7 @@ void setSignalHandler(sighandler_t func)
   setSignalHandler(SIGINT, func);
   setSignalHandler(SIGHUP, func);
   setSignalHandler(SIGTERM, func);
+  signal(SIGPIPE, SIG_IGN);
 }
 
 void blockDefaultSignals(sigset_t *signal_set, sigset_t *old_set)

src/observer/storage/trx/mvcc_trx.cpp

@@ -267,7 +267,8 @@ RC MvccTrx::commit_with_trx_id(int32_t commit_xid)
         trx_fields(table, begin_xid_field, end_xid_field);
 
         auto record_updater = [ this, &begin_xid_field, commit_xid](Record &record) {
-          (void)this;
+          LOG_DEBUG("before commit insert record. trx id=%d, begin xid=%d, commit xid=%d, lbt=%s",
+                    trx_id_, begin_xid_field.get_int(record), commit_xid, lbt());
           ASSERT(begin_xid_field.get_int(record) == -this->trx_id_, 
                  "got an invalid record while committing. begin xid=%d, this trx id=%d", 
                  begin_xid_field.get_int(record), trx_id_);

test/sysbench/miniob_common.lua

+-- Copyright (c) 2023 OceanBase and/or its affiliates. All rights reserved.
+-- miniob is licensed under Mulan PSL v2.
+-- You can use this software according to the terms and conditions of the Mulan PSL v2.
+-- You may obtain a copy of Mulan PSL v2 at:
+--         http://license.coscl.org.cn/MulanPSL2
+-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+-- See the Mulan PSL v2 for more details.
+
+-- -----------------------------------------------------------------------------
+-- Common code for OLTP benchmarks.
+-- -----------------------------------------------------------------------------
+
+function init()
+   assert(event ~= nil,
+          "this script is meant to be included by other OLTP scripts and " ..
+             "should not be called directly.")
+end
+
+if sysbench.cmdline.command == nil then
+   error("Command is required. Supported commands: prepare, warmup, run, " ..
+            "cleanup, help")
+end
+
+-- Command line options
+sysbench.cmdline.options = {
+   table_size =
+      {"Number of rows per table", 10000},
+   range_size =
+      {"Range size for range SELECT queries", 100},
+   tables =
+      {"Number of tables", 1},
+   point_selects =
+      {"Number of point SELECT queries per transaction", 10},
+   simple_ranges =
+      {"Number of simple range SELECT queries per transaction", 1},
+   sum_ranges =
+      {"Number of SELECT SUM() queries per transaction", 1},
+   order_ranges =
+      {"Number of SELECT ORDER BY queries per transaction", 1},
+   distinct_ranges =
+      {"Number of SELECT DISTINCT queries per transaction", 1},
+   index_updates =
+      {"Number of UPDATE index queries per transaction", 1},
+   non_index_updates =
+      {"Number of UPDATE non-index queries per transaction", 1},
+   delete_inserts =
+      {"Number of DELETE/INSERT combinations per transaction", 1},
+   range_selects =
+      {"Enable/disable all range SELECT queries", true},
+   auto_inc =
+   {"Use AUTO_INCREMENT column as Primary Key (for MySQL), " ..
+       "or its alternatives in other DBMS. When disabled, use " ..
+       "client-generated IDs", false},
+   create_table_options =
+      {"Extra CREATE TABLE options", ""},
+   skip_trx =
+      {"Don't start explicit transactions and execute all queries " ..
+          "in the AUTOCOMMIT mode", false},
+   secondary =
+      {"Use a secondary index in place of the PRIMARY KEY", false},
+   create_secondary =
+      {"Create a secondary index in addition to the PRIMARY KEY", true},
+   reconnect =
+      {"Reconnect after every N events. The default (0) is to not reconnect",
+       0},
+   mysql_storage_engine =
+      {"Storage engine, if MySQL is used", "miniob"},
+   pgsql_variant =
+      {"Use this PostgreSQL variant when running with the " ..
+          "PostgreSQL driver. The only currently supported " ..
+          "variant is 'redshift'. When enabled, " ..
+          "create_secondary is automatically disabled, and " ..
+          "delete_inserts is set to 0"}
+}
+
+-- Prepare the dataset. This command supports parallel execution, i.e. will
+-- benefit from executing with --threads > 1 as long as --tables > 1
+function cmd_prepare()
+   local drv = sysbench.sql.driver()
+   local con = drv:connect()
+
+   for i = sysbench.tid % sysbench.opt.threads + 1, sysbench.opt.tables,
+   sysbench.opt.threads do
+     create_table(drv, con, i)
+   end
+end
+
+-- Preload the dataset into the server cache. This command supports parallel
+-- execution, i.e. will benefit from executing with --threads > 1 as long as
+-- --tables > 1
+--
+-- PS. Currently, this command is only meaningful for MySQL/InnoDB benchmarks
+function cmd_warmup()
+   local drv = sysbench.sql.driver()
+   local con = drv:connect()
+
+   assert(drv:name() == "mysql", "warmup is currently MySQL only")
+
+   -- Do not create on disk tables for subsequent queries
+   con:query("SET tmp_table_size=2*1024*1024*1024")
+   con:query("SET max_heap_table_size=2*1024*1024*1024")
+
+   for i = sysbench.tid % sysbench.opt.threads + 1, sysbench.opt.tables,
+   sysbench.opt.threads do
+      local t = "sbtest" .. i
+      print("Preloading table " .. t)
+      con:query("ANALYZE TABLE sbtest" .. i)
+      con:query(string.format(
+                   "SELECT AVG(id) FROM " ..
+                      "(SELECT * FROM %s FORCE KEY (PRIMARY) " ..
+                      "LIMIT %u) t",
+                   t, sysbench.opt.table_size))
+      con:query(string.format(
+                   "SELECT COUNT(*) FROM " ..
+                      "(SELECT * FROM %s WHERE k LIKE '%%0%%' LIMIT %u) t",
+                   t, sysbench.opt.table_size))
+   end
+end
+
+-- Implement parallel prepare and warmup commands, define 'prewarm' as an alias
+-- for 'warmup'
+sysbench.cmdline.commands = {
+   prepare = {cmd_prepare, sysbench.cmdline.PARALLEL_COMMAND},
+   warmup = {cmd_warmup, sysbench.cmdline.PARALLEL_COMMAND},
+   prewarm = {cmd_warmup, sysbench.cmdline.PARALLEL_COMMAND}
+}
+
+
+-- Template strings of random digits with 11-digit groups separated by dashes
+
+-- 10 groups, 119 characters
+local c_value_template = "###########-###########-###########-" ..
+   "###########-###########-###########-" ..
+   "###########-###########-###########-" ..
+   "###########"
+
+-- 5 groups, 59 characters
+local pad_value_template = "###########-###########-###########-" ..
+   "###########-###########"
+
+function get_c_value()
+   return sysbench.rand.string(c_value_template)
+end
+
+function get_pad_value()
+   return sysbench.rand.string(pad_value_template)
+end
+
+function get_f_value()
+   return 1+sysbench.opt.table_size*sysbench.rand.uniform_double()
+end
+
+function create_table(drv, con, table_num)
+   local id_index_def, id_def
+   local extra_table_options = ""
+   local query
+
+   print(string.format("Creating table 'sbtest%d'...", table_num))
+
+   query = string.format([[
+CREATE TABLE sbtest%d(
+  k INT,
+  f FLOAT,
+  c CHAR(120),
+  pad CHAR(60) 
+) %s]],
+      table_num,
+      sysbench.opt.create_table_options)
+
+   con:query(query)
+
+   if (sysbench.opt.table_size > 0) then
+      print(string.format("Inserting %d records into 'sbtest%d'",
+                          sysbench.opt.table_size, table_num))
+   end
+
+   -- query = "INSERT INTO sbtest" .. table_num .. "(k, f, c, pad) VALUES"
+   query = "INSERT INTO sbtest" .. table_num .. " VALUES"
+
+   -- con:bulk_insert_init(query)
+
+   local c_val
+   local pad_val
+   local f_val
+
+   for i = 1, sysbench.opt.table_size do
+
+      c_val = get_c_value()
+      pad_val = get_pad_value()
+      f_val = get_f_value()
+
+      insert_sql = string.format("%s(%d, %f, '%s', '%s')",
+                              query, sysbench.rand.default(1, sysbench.opt.table_size),
+                              f_val, c_val, pad_val)    
+      con:query(insert_sql)
+
+      -- con:bulk_insert_next(query)
+   end
+
+   -- con:bulk_insert_done()
+
+   if sysbench.opt.create_secondary then
+      print(string.format("Creating a secondary index on 'sbtest%d'...",
+                          table_num))
+      con:query(string.format("CREATE INDEX k_%d ON sbtest%d(k)",
+                              table_num, table_num))
+   end
+end
+
+local t = sysbench.sql.type
+local stmt_defs = {
+   point_selects = {
+      "SELECT c FROM sbtest%u WHERE id=?",
+      t.INT},
+   simple_ranges = {
+      "SELECT c FROM sbtest%u WHERE id BETWEEN ? AND ?",
+      t.INT, t.INT},
+   sum_ranges = {
+      "SELECT SUM(k) FROM sbtest%u WHERE id BETWEEN ? AND ?",
+       t.INT, t.INT},
+   order_ranges = {
+      "SELECT c FROM sbtest%u WHERE id BETWEEN ? AND ? ORDER BY c",
+       t.INT, t.INT},
+   distinct_ranges = {
+      "SELECT DISTINCT c FROM sbtest%u WHERE id BETWEEN ? AND ? ORDER BY c",
+      t.INT, t.INT},
+   index_updates = {
+      "UPDATE sbtest%u SET k=k+1 WHERE id=?",
+      t.INT},
+   non_index_updates = {
+      "UPDATE sbtest%u SET c=? WHERE id=?",
+      {t.CHAR, 120}, t.INT},
+   deletes = {
+      "DELETE FROM sbtest%u WHERE id=?",
+      t.INT},
+   inserts = {
+      "INSERT INTO sbtest%u (id, k, c, pad) VALUES (?, ?, ?, ?)",
+      t.INT, t.INT, {t.CHAR, 120}, {t.CHAR, 60}},
+}
+
+function prepare_begin()
+   stmt.begin = con:prepare("BEGIN")
+end
+
+function prepare_commit()
+   stmt.commit = con:prepare("COMMIT")
+end
+
+function prepare_for_each_table(key)
+   for t = 1, sysbench.opt.tables do
+      stmt[t][key] = con:prepare(string.format(stmt_defs[key][1], t))
+
+      local nparam = #stmt_defs[key] - 1
+
+      if nparam > 0 then
+         param[t][key] = {}
+      end
+
+      for p = 1, nparam do
+         local btype = stmt_defs[key][p+1]
+         local len
+
+         if type(btype) == "table" then
+            len = btype[2]
+            btype = btype[1]
+         end
+         if btype == sysbench.sql.type.VARCHAR or
+            btype == sysbench.sql.type.CHAR then
+               param[t][key][p] = stmt[t][key]:bind_create(btype, len)
+         else
+            param[t][key][p] = stmt[t][key]:bind_create(btype)
+         end
+      end
+
+      if nparam > 0 then
+         stmt[t][key]:bind_param(unpack(param[t][key]))
+      end
+   end
+end
+
+function prepare_point_selects()
+   prepare_for_each_table("point_selects")
+end
+
+function prepare_simple_ranges()
+   prepare_for_each_table("simple_ranges")
+end
+
+function prepare_sum_ranges()
+   prepare_for_each_table("sum_ranges")
+end
+
+function prepare_order_ranges()
+   prepare_for_each_table("order_ranges")
+end
+
+function prepare_distinct_ranges()
+   prepare_for_each_table("distinct_ranges")
+end
+
+function prepare_index_updates()
+   prepare_for_each_table("index_updates")
+end
+
+function prepare_non_index_updates()
+   prepare_for_each_table("non_index_updates")
+end
+
+function prepare_delete_inserts()
+   prepare_for_each_table("deletes")
+   prepare_for_each_table("inserts")
+end
+
+function thread_init()
+   drv = sysbench.sql.driver()
+   con = drv:connect()
+
+   -- Create global nested tables for prepared statements and their
+   -- parameters. We need a statement and a parameter set for each combination
+   -- of connection/table/query
+   stmt = {}
+   param = {}
+
+   for t = 1, sysbench.opt.tables do
+      stmt[t] = {}
+      param[t] = {}
+   end
+
+   -- This function is a 'callback' defined by individual benchmark scripts
+   prepare_statements()
+end
+
+-- Close prepared statements
+function close_statements()
+   for t = 1, sysbench.opt.tables do
+      for k, s in pairs(stmt[t]) do
+         stmt[t][k]:close()
+      end
+   end
+   if (stmt.begin ~= nil) then
+      stmt.begin:close()
+   end
+   if (stmt.commit ~= nil) then
+      stmt.commit:close()
+   end
+end
+
+function thread_done()
+   close_statements()
+   con:disconnect()
+end
+
+function cleanup()
+   local drv = sysbench.sql.driver()
+   local con = drv:connect()
+
+   for i = 1, sysbench.opt.tables do
+      print(string.format("Dropping table 'sbtest%d'...", i))
+      con:query("DROP TABLE IF EXISTS sbtest" .. i )
+   end
+end
+
+local function get_table_num()
+   return sysbench.rand.uniform(1, sysbench.opt.tables)
+end
+
+local function get_id()
+   return sysbench.rand.default(1, sysbench.opt.table_size)
+end
+
+function begin()
+   stmt.begin:execute()
+end
+
+function commit()
+   stmt.commit:execute()
+end
+
+function execute_point_selects()
+   local tnum = get_table_num()
+   local i
+
+   for i = 1, sysbench.opt.point_selects do
+      param[tnum].point_selects[1]:set(get_id())
+
+      stmt[tnum].point_selects:execute()
+   end
+end
+
+local function execute_range(key)
+   local tnum = get_table_num()
+
+   for i = 1, sysbench.opt[key] do
+      local id = get_id()
+
+      param[tnum][key][1]:set(id)
+      param[tnum][key][2]:set(id + sysbench.opt.range_size - 1)
+
+      stmt[tnum][key]:execute()
+   end
+end
+
+function execute_simple_ranges()
+   execute_range("simple_ranges")
+end
+
+function execute_sum_ranges()
+   execute_range("sum_ranges")
+end
+
+function execute_order_ranges()
+   execute_range("order_ranges")
+end
+
+function execute_distinct_ranges()
+   execute_range("distinct_ranges")
+end
+
+function execute_index_updates()
+   local tnum = get_table_num()
+
+   for i = 1, sysbench.opt.index_updates do
+      param[tnum].index_updates[1]:set(get_id())
+
+      stmt[tnum].index_updates:execute()
+   end
+end
+
+function execute_non_index_updates()
+   local tnum = get_table_num()
+
+   for i = 1, sysbench.opt.non_index_updates do
+      param[tnum].non_index_updates[1]:set_rand_str(c_value_template)
+      param[tnum].non_index_updates[2]:set(get_id())
+
+      stmt[tnum].non_index_updates:execute()
+   end
+end
+
+function execute_delete_inserts()
+   local tnum = get_table_num()
+
+   for i = 1, sysbench.opt.delete_inserts do
+      local id = get_id()
+      local k = get_id()
+
+      param[tnum].deletes[1]:set(id)
+
+      param[tnum].inserts[1]:set(id)
+      param[tnum].inserts[2]:set(k)
+      param[tnum].inserts[3]:set_rand_str(c_value_template)
+      param[tnum].inserts[4]:set_rand_str(pad_value_template)
+
+      stmt[tnum].deletes:execute()
+      stmt[tnum].inserts:execute()
+   end
+end
+
+-- Re-prepare statements if we have reconnected, which is possible when some of
+-- the listed error codes are in the --mysql-ignore-errors list
+function sysbench.hooks.before_restart_event(errdesc)
+   if errdesc.sql_errno == 2013 or -- CR_SERVER_LOST
+      errdesc.sql_errno == 2055 or -- CR_SERVER_LOST_EXTENDED
+      errdesc.sql_errno == 2006 or -- CR_SERVER_GONE_ERROR
+      errdesc.sql_errno == 2011    -- CR_TCP_CONNECTION
+   then
+      close_statements()
+      prepare_statements()
+   end
+end
+
+function check_reconnect()
+   if sysbench.opt.reconnect > 0 then
+      transactions = (transactions or 0) + 1
+      if transactions % sysbench.opt.reconnect == 0 then
+         close_statements()
+         con:reconnect()
+         prepare_statements()
+      end
+   end
+end
+

test/sysbench/miniob_insert.lua

+#!/usr/bin/env sysbench
+-- Copyright (c) 2023 OceanBase and/or its affiliates. All rights reserved.
+-- miniob is licensed under Mulan PSL v2.
+-- You can use this software according to the terms and conditions of the Mulan PSL v2.
+-- You may obtain a copy of Mulan PSL v2 at:
+--         http://license.coscl.org.cn/MulanPSL2
+-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+-- See the Mulan PSL v2 for more details.
+
+-- ----------------------------------------------------------------------
+-- Insert-Only OLTP benchmark
+-- ----------------------------------------------------------------------
+
+require("miniob_common")
+
+sysbench.cmdline.commands.prepare = {
+   function ()
+      cmd_prepare()
+   end,
+   sysbench.cmdline.PARALLEL_COMMAND
+}
+
+function prepare_statements()
+   -- We do not use prepared statements here, but oltp_common.sh expects this
+   -- function to be defined
+end
+
+function event()
+   local table_name = "sbtest" .. sysbench.rand.uniform(1, sysbench.opt.tables)
+   local k_val = sysbench.rand.default(1, sysbench.opt.table_size)
+   local c_val = get_c_value()
+   local pad_val = get_pad_value()
+   local f_val = get_f_value()
+
+   con:query(string.format("INSERT INTO %s VALUES " ..
+                              "(%d, %f, '%s', '%s')",
+                           table_name, k_val, f_val, c_val, pad_val))
+
+
+   check_reconnect()
+end
+