benchmark: remove dependency of unexport rocksdb objects (#372)

run.sh

@@ -1491,30 +1491,49 @@ function run_clear_upgrade_test()
 function usage_bench()
 {
     echo "Options for subcommand 'bench':"
-    echo "   -h|--help            print the help info"
-    echo "   -t|--type            benchmark type, supporting:"
-    echo "                          fillseq_pegasus, fillrandom_pegasus, readrandom_pegasus, filluniquerandom_pegasus,"
-    echo "                          deleteseq_pegasus,deleterandom_pegasus,multi_set_pegasus,scan_pegasus"
-    echo "                        default is fillseq_pegasus,readrandom_pegasus"
-    echo "   -n <num>             number of key/value pairs, default 100000"
-    echo "   --cluster <str>      cluster meta lists, default '127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603'"
-    echo "   --app_name <str>     app name, default 'temp'"
-    echo "   --thread_num <num>   number of threads, default 1"
-    echo "   --key_size <num>     key size, default 16"
-    echo "   --value_size <num>   value size, default 100"
-    echo "   --timeout <num>      timeout in milliseconds, default 1000"
+    echo "   -h|--help                 print the help info"
+    echo "   --type                    benchmark type, supporting:"
+    echo "                             fillrandom_pegasus       --pegasus write N random values with random keys list"
+    echo "                             readrandom_pegasus       --pegasus read N times with random keys list"
+    echo "                             deleterandom_pegasus     --pegasus delete N entries with random keys list"
+    echo "                             Comma-separated list of operations is going to run in the specified order."
+    echo "                             default is 'fillrandom_pegasus,readrandom_pegasus,deleterandom_pegasus'"
+    echo "   --num <num>               number of key/value pairs, default is 10000"
+    echo "   --cluster <str>           cluster meta lists, default is '127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603'"
+    echo "   --app_name <str>          app name, default is 'temp'"
+    echo "   --thread_num <num>        number of threads, default is 1"
+    echo "   --hashkey_size <num>      hashkey size in bytes, default is 16"
+    echo "   --sortkey_size <num>      sortkey size in bytes, default is 16"
+    echo "   --value_size <num>        value size in bytes, default is 100"
+    echo "   --timeout <num>           timeout in milliseconds, default is 1000"
+    echo "   --seed <num>              seed base for random number generator, When 0 it is specified as 1000. default is 1000"
+}
+
+function fill_bench_config() {
+    sed -i "s/@TYPE@/$TYPE/g" ./config-bench.ini
+    sed -i "s/@NUM@/$NUM/g" ./config-bench.ini
+    sed -i "s/@CLUSTER@/$CLUSTER/g" ./config-bench.ini
+    sed -i "s/@APP@/$APP/g" ./config-bench.ini
+    sed -i "s/@THREAD@/$THREAD/g" ./config-bench.ini
+    sed -i "s/@HASHKEY_SIZE@/$HASHKEY_SIZE/g" ./config-bench.ini
+    sed -i "s/@SORTKEY_SIZE@/$SORTKEY_SIZE/g" ./config-bench.ini
+    sed -i "s/@VALUE_SIZE@/$VALUE_SIZE/g" ./config-bench.ini
+    sed -i "s/@TIMEOUT_MS@/$TIMEOUT_MS/g" ./config-bench.ini
+    sed -i "s/@SEED@/$SEED/g" ./config-bench.ini
 }
 
 function run_bench()
 {
-    TYPE=fillseq_pegasus,readrandom_pegasus
-    NUM=100000
+    TYPE=fillrandom_pegasus,readrandom_pegasus,deleterandom_pegasus
+    NUM=10000
     CLUSTER=127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603
     APP=temp
     THREAD=1
-    KEY_SIZE=16
+    HASHKEY_SIZE=16
+    SORTKEY_SIZE=16
     VALUE_SIZE=100
     TIMEOUT_MS=1000
+    SEED=1000
     while [[ $# > 0 ]]; do
         key="$1"
         case $key in
@@ -1522,11 +1541,11 @@ function run_bench()
                 usage_bench
                 exit 0
                 ;;
-            -t|--type)
+            --type)
                 TYPE="$2"
                 shift
                 ;;
-            -n)
+            --num)
                 NUM="$2"
                 shift
                 ;;
@@ -1542,8 +1561,12 @@ function run_bench()
                 THREAD="$2"
                 shift
                 ;;
-            --key_size)
-                KEY_SIZE="$2"
+            --hashkey_size)
+                HASHKEY_SIZE="$2"
+                shift
+                ;;
+            --sortkey_size)
+                SORTKEY_SIZE="$2"
                 shift
                 ;;
             --value_size)
@@ -1554,6 +1577,10 @@ function run_bench()
                 TIMEOUT_MS="$2"
                 shift
                 ;;
+            --seed)
+                SEED="$2"
+                shift
+                ;;
             *)
                 echo "ERROR: unknown option \"$key\""
                 echo
@@ -1563,14 +1590,12 @@ function run_bench()
         esac
         shift
     done
-
     cd ${ROOT}
-    sed -i "s/@CLUSTER@/$CLUSTER/g" ${DSN_ROOT}/bin/pegasus_bench/config.ini
+    cp ${DSN_ROOT}/bin/pegasus_bench/config.ini ./config-bench.ini
+    fill_bench_config
     ln -s -f ${DSN_ROOT}/bin/pegasus_bench/pegasus_bench
-    ./pegasus_bench --pegasus_config=${DSN_ROOT}/bin/pegasus_bench/config.ini --benchmarks=${TYPE} --pegasus_timeout_ms=${TIMEOUT_MS} \
-        --key_size=${KEY_SIZE} --value_size=${VALUE_SIZE} --threads=${THREAD} --num=${NUM} \
-        --pegasus_cluster_name=mycluster --pegasus_app_name=${APP} --stats_interval=1000 --histogram=1 \
-        --compression_ratio=1.0
+    ./pegasus_bench ./config-bench.ini
+    rm -f ./config-bench.ini
 }
 
 #####################

src/test/bench_test/CMakeLists.txt

@@ -15,7 +15,7 @@ set(MY_PROJ_INC_PATH "../../../rocksdb")
 set(MY_PROJ_LIBS
         pegasus_client_static
         RocksDB::rocksdb
-        gflags)
+        fmt)
 
 set(MY_BOOST_LIBS Boost::system Boost::filesystem)
 

src/test/bench_test/benchmark.cpp

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <sstream>
+#include <pegasus/client.h>
+#include <dsn/dist/fmt_logging.h>
+
+#include "benchmark.h"
+#include "rand.h"
+
+namespace pegasus {
+namespace test {
+benchmark::benchmark()
+{
+    _client = pegasus_client_factory::get_client(config::instance().pegasus_cluster_name.c_str(),
+                                                 config::instance().pegasus_app_name.c_str());
+    assert(nullptr != _client);
+
+    // init operation method map
+    _operation_method = {{kUnknown, nullptr},
+                         {kRead, &benchmark::read_random},
+                         {kWrite, &benchmark::write_random},
+                         {kDelete, &benchmark::delete_random}};
+}
+
+void benchmark::run()
+{
+    // print summarize information
+    print_header();
+
+    std::stringstream benchmark_stream(config::instance().benchmarks);
+    std::string name;
+    while (std::getline(benchmark_stream, name, ',')) {
+        // run the specified benchmark
+        operation_type op_type = get_operation_type(name);
+        run_benchmark(config::instance().threads, op_type);
+    }
+}
+
+void benchmark::run_benchmark(int thread_count, operation_type op_type)
+{
+    // get method by operation type
+    bench_method method = _operation_method[op_type];
+    assert(method != nullptr);
+
+    // create histogram statistic
+    std::shared_ptr<rocksdb::Statistics> hist_stats = rocksdb::CreateDBStatistics();
+
+    // create thread args for each thread, and run them
+    std::vector<std::shared_ptr<thread_arg>> args;
+    for (int i = 0; i < thread_count; i++) {
+        args.push_back(
+            std::make_shared<thread_arg>(i + config::instance().seed, hist_stats, method, this));
+        config::instance().env->StartThread(thread_body, args[i].get());
+    }
+
+    // wait all threads are done
+    config::instance().env->WaitForJoin();
+
+    // merge statistics
+    statistics merge_stats(hist_stats);
+    for (int i = 0; i < thread_count; i++) {
+        merge_stats.merge(args[i]->stats);
+    }
+    merge_stats.report(op_type);
+}
+
+void benchmark::thread_body(void *v)
+{
+    thread_arg *arg = reinterpret_cast<thread_arg *>(v);
+
+    // reseed local random generator
+    reseed_thread_local_rng(arg->seed);
+
+    // progress the method
+    arg->stats.start();
+    (arg->bm->*(arg->method))(arg);
+    arg->stats.stop();
+}
+
+void benchmark::write_random(thread_arg *thread)
+{
+    // do write operation num times
+    uint64_t bytes = 0;
+    int count = 0;
+    for (int i = 0; i < config::instance().num; i++) {
+        // generate hash key and sort key
+        std::string hashkey, sortkey, value;
+        generate_kv_pair(hashkey, sortkey, value);
+
+        // write to pegasus
+        int try_count = 0;
+        while (true) {
+            try_count++;
+            int ret = _client->set(hashkey, sortkey, value, config::instance().pegasus_timeout_ms);
+            if (ret == ::pegasus::PERR_OK) {
+                bytes += config::instance().value_size + config::instance().hashkey_size +
+                         config::instance().sortkey_size;
+                count++;
+                break;
+            } else if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Set returned an error: {}\n", _client->get_error_string(ret));
+                exit(1);
+            } else {
+                fmt::print(stderr, "Set timeout, retry({})\n", try_count);
+            }
+        }
+
+        // count this operation
+        thread->stats.finished_ops(1, kWrite);
+    }
+
+    // count total write bytes
+    thread->stats.add_bytes(bytes);
+}
+
+void benchmark::read_random(thread_arg *thread)
+{
+    uint64_t bytes = 0;
+    uint64_t found = 0;
+    for (int i = 0; i < config::instance().num; i++) {
+        // generate hash key and sort key
+        // generate value for random to keep in peace with write
+        std::string hashkey, sortkey, value;
+        generate_kv_pair(hashkey, sortkey, value);
+
+        // read from pegasus
+        int try_count = 0;
+        while (true) {
+            try_count++;
+            int ret = _client->get(hashkey, sortkey, value, config::instance().pegasus_timeout_ms);
+            if (ret == ::pegasus::PERR_OK) {
+                found++;
+                bytes += hashkey.size() + sortkey.size() + value.size();
+                break;
+            } else if (ret == ::pegasus::PERR_NOT_FOUND) {
+                break;
+            } else if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Get returned an error: {}\n", _client->get_error_string(ret));
+                exit(1);
+            } else {
+                fmt::print(stderr, "Get timeout, retry({})\n", try_count);
+            }
+        }
+
+        // count this operation
+        thread->stats.finished_ops(1, kRead);
+    }
+
+    // count total read bytes and hit rate
+    std::string msg = fmt::format("({} of {} found)", found, config::instance().num);
+    thread->stats.add_bytes(bytes);
+    thread->stats.add_message(msg);
+}
+
+void benchmark::delete_random(thread_arg *thread)
+{
+    // do delete operation num times
+    for (int i = 0; i < config::instance().num; i++) {
+        // generate hash key and sort key
+        // generate value for random to keep in peace with write
+        std::string hashkey, sortkey, value;
+        generate_kv_pair(hashkey, sortkey, value);
+
+        int try_count = 0;
+        while (true) {
+            try_count++;
+            int ret = _client->del(hashkey, sortkey, config::instance().pegasus_timeout_ms);
+            if (ret == ::pegasus::PERR_OK) {
+                break;
+            } else if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Del returned an error: {}\n", _client->get_error_string(ret));
+                exit(1);
+            } else {
+                fmt::print(stderr, "Get timeout, retry({})\n", try_count);
+            }
+        }
+
+        // count this operation
+        thread->stats.finished_ops(1, kDelete);
+    }
+}
+
+void benchmark::generate_kv_pair(std::string &hashkey, std::string &sortkey, std::string &value)
+{
+    hashkey = generate_string(config::instance().hashkey_size);
+    sortkey = generate_string(config::instance().sortkey_size);
+    value = generate_string(config::instance().value_size);
+}
+
+operation_type benchmark::get_operation_type(const std::string &name)
+{
+    operation_type op_type = kUnknown;
+    if (name == "fillrandom_pegasus") {
+        op_type = kWrite;
+    } else if (name == "readrandom_pegasus") {
+        op_type = kRead;
+    } else if (name == "deleterandom_pegasus") {
+        op_type = kDelete;
+    } else if (!name.empty()) { // No error message for empty name
+        fmt::print(stderr, "unknown benchmark '{}'\n", name);
+        exit(1);
+    }
+
+    return op_type;
+}
+
+void benchmark::print_header()
+{
+    const config &config_ = config::instance();
+    fmt::print(stdout, "Hashkeys:       {} bytes each\n", config_.hashkey_size);
+    fmt::print(stdout, "Sortkeys:       {} bytes each\n", config_.sortkey_size);
+    fmt::print(stdout, "Values:         {} bytes each\n", config_.value_size);
+    fmt::print(stdout, "Entries:        {}\n", config_.num);
+    fmt::print(stdout,
+               "FileSize:       {} MB (estimated)\n",
+               ((config_.hashkey_size + config_.sortkey_size + config_.value_size) * config_.num) >>
+                   20);
+
+    print_warnings();
+    fmt::print(stdout, "------------------------------------------------\n");
+}
+
+void benchmark::print_warnings()
+{
+#if defined(__GNUC__) && !defined(__OPTIMIZE__)
+    fmt::print(stdout, "WARNING: Optimization is disabled: benchmarks unnecessarily slow\n");
+#endif
+#ifndef NDEBUG
+    fmt::print(stdout, "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n");
+#endif
+}
+} // namespace test
+} // namespace pegasus

src/test/bench_test/benchmark.h

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <unordered_map>
+
+#include "statistics.h"
+#include "config.h"
+
+namespace pegasus {
+namespace test {
+
+class benchmark;
+struct thread_arg;
+typedef void (benchmark::*bench_method)(thread_arg *);
+
+struct thread_arg
+{
+    int64_t seed;
+    statistics stats;
+    bench_method method;
+    benchmark *bm;
+
+    thread_arg(uint64_t seed_,
+               std::shared_ptr<rocksdb::Statistics> hist_stats_,
+               bench_method bench_method_,
+               benchmark *benchmark_)
+        : seed(seed_), stats(hist_stats_), method(bench_method_), bm(benchmark_)
+    {
+    }
+};
+
+class benchmark
+{
+public:
+    benchmark();
+    ~benchmark() = default;
+    void run();
+
+private:
+    /** thread main function */
+    static void thread_body(void *v);
+
+    /** benchmark operations **/
+    void run_benchmark(int thread_count, operation_type op_type);
+    void write_random(thread_arg *thread);
+    void read_random(thread_arg *thread);
+    void delete_random(thread_arg *thread);
+
+    /**  generate hash/sort key and value */
+    void generate_kv_pair(std::string &hashkey, std::string &sortkey, std::string &value);
+
+    /** some auxiliary functions */
+    operation_type get_operation_type(const std::string &name);
+    void print_header();
+    void print_warnings();
+
+private:
+    // the pegasus client to do read/write/delete operations
+    pegasus_client *_client;
+    // the map of operation type and the process method
+    std::unordered_map<operation_type, bench_method, std::hash<unsigned char>> _operation_method;
+};
+} // namespace test
+} // namespace pegasus

src/test/bench_test/config.cpp

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <dsn/utility/config_api.h>
+#include "config.h"
+
+namespace pegasus {
+namespace test {
+
+config::config()
+{
+    pegasus_cluster_name = dsn_config_get_value_string(
+        "pegasus.benchmark", "pegasus_cluster_name", "onebox", "pegasus cluster name");
+    pegasus_app_name = dsn_config_get_value_string(
+        "pegasus.benchmark", "pegasus_app_name", "temp", "pegasus app name");
+    pegasus_timeout_ms =
+        (int32_t)dsn_config_get_value_uint64("pegasus.benchmark",
+                                             "pegasus_timeout_ms",
+                                             1000,
+                                             "pegasus read/write timeout in milliseconds");
+    benchmarks = dsn_config_get_value_string(
+        "pegasus.benchmark",
+        "benchmarks",
+        "fillrandom_pegasus,readrandom_pegasus,deleterandom_pegasus",
+        "Comma-separated list of operations to run in the specified order. Available benchmarks:\n"
+        "\tfillrandom_pegasus       -- pegasus write N values in random key order\n"
+        "\treadrandom_pegasus       -- pegasus read N times in random order\n"
+        "\tdeleterandom_pegasus     -- pegasus delete N keys in random order\n");
+    num = dsn_config_get_value_uint64(
+        "pegasus.benchmark", "num", 10000, "Number of key/values to place in database");
+    threads = (int32_t)dsn_config_get_value_uint64(
+        "pegasus.benchmark", "threads", 1, "Number of concurrent threads to run");
+    hashkey_size = (int32_t)dsn_config_get_value_uint64(
+        "pegasus.benchmark", "hashkey_size", 16, "size of each hashkey");
+    sortkey_size = (int32_t)dsn_config_get_value_uint64(
+        "pegasus.benchmark", "sortkey_size", 16, "size of each sortkey");
+    value_size = (int32_t)dsn_config_get_value_uint64(
+        "pegasus.benchmark", "value_size", 100, "Size of each value");
+    seed = dsn_config_get_value_uint64(
+        "pegasus.benchmark",
+        "seed",
+        1000,
+        "Seed base for random number generators. When 0 it is deterministic");
+    seed = seed ? seed : 1000;
+    env = rocksdb::Env::Default();
+}
+} // namespace test
+} // namespace pegasus

src/test/bench_test/config.h

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <string>
+#include <rocksdb/env.h>
+#include <dsn/utility/singleton.h>
+
+namespace pegasus {
+namespace test {
+/** Thread safety singleton */
+struct config : public ::dsn::utils::singleton<config>
+{
+    config();
+
+    std::string pegasus_cluster_name;
+    std::string pegasus_app_name;
+    // Pegasus read/write/delete timeout in milliseconds
+    uint32_t pegasus_timeout_ms;
+    // Comma-separated list of operations to run
+    std::string benchmarks;
+    // Number of key/values to place in database
+    uint64_t num;
+    // Number of concurrent threads to run
+    uint32_t threads;
+    // size of each value
+    uint32_t value_size;
+    // size of each hashkey
+    uint32_t hashkey_size;
+    // size of each sortkey
+    uint32_t sortkey_size;
+    // Seed base for random number generators
+    uint64_t seed;
+    // Default environment suitable for the current operating system
+    rocksdb::Env *env;
+};
+} // namespace test
+} // namespace pegasus

src/test/bench_test/config.ini

@@ -54,4 +54,17 @@ worker_priority = THREAD_xPRIORITY_NORMAL
 worker_count = 10
 
 [pegasus.clusters]
-mycluster = @CLUSTER@
+onebox = @CLUSTER@
+
+[pegasus.benchmark]
+pegasus_cluster_name = onebox
+pegasus_app_name = @APP@
+pegasus_timeout_ms = @TIMEOUT_MS@
+benchmarks = @TYPE@
+num = @NUM@
+threads = @THREAD@
+value_size = @VALUE_SIZE@
+hashkey_size = @HASHKEY_SIZE@
+sortkey_size = @SORTKEY_SIZE@
+seed = @SEED@
+

src/test/bench_test/main.cpp

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <unistd.h>
+#include <pegasus/client.h>
+#include <dsn/dist/fmt_logging.h>
+
+#include "benchmark.h"
+
+int db_bench_tool(const char *config_file)
+{
+    bool init = pegasus::pegasus_client_factory::initialize(config_file);
+    if (!init) {
+        fmt::print(stderr, "Init pegasus error\n");
+        return -1;
+    }
+    sleep(1);
+    fmt::print(stdout, "Init pegasus succeed\n");
+
+    pegasus::test::benchmark bm;
+    bm.run();
+    sleep(1); // Sleep a while to exit gracefully.
+
+    return 0;
+}
+
+int main(int argc, char **argv)
+{
+    if (argc < 2) {
+        fmt::print(stderr, "USAGE: {} <config-file>", argv[0]);
+        return -1;
+    }
+
+    return db_bench_tool(argv[1]);
+}

src/test/bench_test/rand.cpp

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <random>
+
+namespace pegasus {
+namespace test {
+thread_local std::ranlux48_base thread_local_rng(std::random_device{}());
+
+void reseed_thread_local_rng(uint64_t seed) { thread_local_rng.seed(seed); }
+
+uint64_t next_u64()
+{
+    return std::uniform_int_distribution<uint64_t>(0, std::numeric_limits<uint64_t>::max())(
+        thread_local_rng);
+}
+
+std::string generate_string(uint64_t len)
+{
+    std::string key;
+
+    // fill with random int
+    uint64_t random_int = next_u64();
+    key.append(reinterpret_cast<char *>(&random_int), std::min(len, 8UL));
+
+    // append with '0'
+    key.resize(len, '0');
+
+    return key;
+}
+} // namespace test
+} // namespace pegasus

src/test/bench_test/rand.h

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+namespace pegasus {
+namespace test {
+// Reseeds the RNG of current thread.
+extern void reseed_thread_local_rng(uint64_t seed);
+extern uint64_t next_u64();
+extern std::string generate_string(uint64_t len);
+} // namespace test
+} // namespace pegasus

src/test/bench_test/statistics.cpp

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <unordered_map>
+#include <dsn/dist/fmt_logging.h>
+
+#include "statistics.h"
+#include "config.h"
+
+namespace pegasus {
+namespace test {
+std::unordered_map<operation_type, std::string, std::hash<unsigned char>> operation_type_string = {
+    {kUnknown, "unKnown"}, {kRead, "read"}, {kWrite, "write"}, {kDelete, "delete"}};
+
+statistics::statistics(std::shared_ptr<rocksdb::Statistics> hist_stats)
+{
+    _next_report = 100;
+    _done = 0;
+    _bytes = 0;
+    _start = config::instance().env->NowMicros();
+    _last_op_finish = _start;
+    _finish = _start;
+    _hist_stats = hist_stats;
+    _message.clear();
+}
+
+void statistics::start()
+{
+    _start = config::instance().env->NowMicros();
+    _last_op_finish = _start;
+    _finish = _start;
+}
+
+void statistics::merge(const statistics &other)
+{
+    _done += other._done;
+    _bytes += other._bytes;
+    _start = std::min(other._start, _start);
+    _finish = std::max(other._finish, _finish);
+    this->add_message(other._message);
+}
+
+void statistics::stop() { _finish = config::instance().env->NowMicros(); }
+
+void statistics::finished_ops(int64_t num_ops, enum operation_type op_type)
+{
+    uint64_t now = config::instance().env->NowMicros();
+    uint64_t micros = now - _last_op_finish;
+    _last_op_finish = now;
+
+    // if there is a long operation, print warning message
+    if (micros > 20000) {
+        fmt::print(stderr, "long op: {} micros\r", micros);
+    }
+
+    // print the benchmark running status
+    _done += num_ops;
+    if (_done >= _next_report) {
+        _next_report += report_step(_next_report);
+        fmt::print(stderr, "... finished {} ops\r", _done);
+    }
+
+    // add excution time of this operation to _hist_stats
+    if (_hist_stats) {
+        _hist_stats->measureTime(op_type, micros);
+    }
+}
+
+void statistics::report(operation_type op_type)
+{
+    // Pretend at least one op was done in case we are running a benchmark
+    // that does not call finished_ops().
+    if (_done < 1)
+        _done = 1;
+
+    // elasped time(s)
+    double elapsed = (_finish - _start) * 1e-6;
+
+    // append rate(MBytes) and _message to extra
+    std::string extra;
+    if (_bytes > 0) {
+        // Rate is computed on actual elapsed time, not the sum of per-thread
+        // elapsed times.
+        extra = fmt::format("{} MB/s ", (_bytes >> 20) / elapsed);
+    }
+    extra.append(_message);
+
+    // print report
+    fmt::print(stdout,
+               "Statistics for {}:  \n{} micros/op; {} ops/sec; {}\n",
+               operation_type_string[op_type],
+               elapsed * 1e6 / _done,
+               static_cast<long>(_done / elapsed),
+               extra);
+
+    // print histogram if _hist_stats is not NULL
+    if (_hist_stats) {
+        fmt::print(stdout, "{}\n", _hist_stats->getHistogramString(op_type));
+    }
+}
+
+void statistics::add_message(const std::string &msg)
+{
+    if (msg.empty())
+        return;
+
+    if (!_message.empty()) {
+        _message.push_back(' ');
+    }
+    _message.append(msg);
+}
+
+void statistics::add_bytes(int64_t n) { _bytes += n; }
+
+uint32_t statistics::report_step(uint64_t current_report) const
+{
+    uint32_t step = 0;
+    switch (current_report) {
+    case 0 ... 999:
+        step = 100;
+        break;
+    case 1000 ... 4999:
+        step = 500;
+        break;
+    case 5000 ... 9999:
+        step = 1000;
+        break;
+    case 10000 ... 49999:
+        step = 5000;
+        break;
+    case 50000 ... 99999:
+        step = 10000;
+        break;
+    case 100000 ... 499999:
+        step = 50000;
+        break;
+    default:
+        step = 100000;
+        break;
+    }
+
+    return step;
+}
+} // namespace test
+} // namespace pegasus

src/test/bench_test/statistics.h

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <rocksdb/statistics.h>
+
+#include "utils.h"
+
+namespace pegasus {
+namespace test {
+class statistics
+{
+public:
+    statistics(std::shared_ptr<rocksdb::Statistics> hist_stats);
+    void start();
+    void finished_ops(int64_t num_ops, enum operation_type op_type);
+    void stop();
+    void merge(const statistics &other);
+    void report(operation_type op_type);
+    void add_bytes(int64_t n);
+    void add_message(const std::string &msg);
+
+private:
+    uint32_t report_step(uint64_t current_report) const;
+
+    // thread id which controls this statistics
+    int _tid;
+    // the start time of benchmark
+    uint64_t _start;
+    // the stop time of benchmark
+    uint64_t _finish;
+    // how many operations are done
+    uint64_t _done;
+    // the point(operation count) at which the next report
+    uint64_t _next_report;
+    // how many bytes the benchmark read/write
+    uint64_t _bytes;
+    // the last operation's finish time
+    uint64_t _last_op_finish;
+    // the information of benchmark operation
+    std::string _message;
+    // histogram performance analyzer
+    std::shared_ptr<rocksdb::Statistics> _hist_stats;
+};
+} // namespace test
+} // namespace pegasus

src/test/bench_test/utils.h

+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+namespace pegasus {
+namespace test {
+enum operation_type
+{
+    kUnknown = 0,
+    kRead,
+    kWrite,
+    kDelete
+};
+} // namespace test
+} // namespace pegasus