feat(test): add bench type multi set/get

run.sh

@@ -1399,6 +1399,8 @@ function usage_bench()
     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 "                             multisetrandom_pegasus   --pegasus write N random values with multi_count hash keys list"
+    echo "                             multigetrandom_pegasus   --pegasus read N random keys with multi_count hash 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"
@@ -1410,6 +1412,7 @@ function usage_bench()
     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"
+    echo "   --multi_count <num>       values count of the same hashkey, used by multi_set/multi_get, default is 100"
 }
 
 function fill_bench_config() {
@@ -1423,6 +1426,7 @@ function fill_bench_config() {
     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
+    sed -i "s/@MULTI_COUNT@/$MULTI_COUNT/g" ./config-bench.ini
 }
 
 function run_bench()
@@ -1437,6 +1441,7 @@ function run_bench()
     VALUE_SIZE=100
     TIMEOUT_MS=1000
     SEED=1000
+    MULTI_COUNT=100
     while [[ $# > 0 ]]; do
         key="$1"
         case $key in
@@ -1484,6 +1489,10 @@ function run_bench()
                 SEED="$2"
                 shift
                 ;;
+            --multi_count)
+                MULTI_COUNT="$2"
+                shift
+                ;;
             *)
                 echo "ERROR: unknown option \"$key\""
                 echo

src/test/bench_test/benchmark.cpp

@@ -45,6 +45,7 @@ DSN_DECLARE_int32(pegasus_timeout_ms);
 DSN_DECLARE_int32(sortkey_size);
 DSN_DECLARE_int32(threads);
 DSN_DECLARE_int32(value_size);
+DSN_DECLARE_int32(multi_count);
 
 benchmark::benchmark()
 {
@@ -56,6 +57,8 @@ benchmark::benchmark()
     _operation_method = {{kUnknown, nullptr},
                          {kRead, &benchmark::read_random},
                          {kWrite, &benchmark::write_random},
+                         {kMultiSet, &benchmark::multi_set_random},
+                         {kMultiGet, &benchmark::multi_get_random},
                          {kDelete, &benchmark::delete_random}};
 }
 
@@ -152,6 +155,53 @@ void benchmark::write_random(thread_arg *thread)
     thread->stats.add_bytes(bytes);
 }
 
+void benchmark::multi_set_random(thread_arg *thread)
+{
+    if (FLAGS_benchmark_num % FLAGS_multi_count != 0) {
+        fmt::print(stderr, "num {} should be a multiple of multi_count({}).\n", FLAGS_benchmark_num, FLAGS_multi_count);
+        dsn_exit(1);
+    }
+
+    // do write operation num times
+    uint64_t bytes = 0;
+
+    for (int i = 0; i < FLAGS_benchmark_num / FLAGS_multi_count; i++) {
+        // generate hash key
+        std::string hash_key, sort_key, value;
+        hash_key = generate_string(FLAGS_hashkey_size);
+        std::map<std::string, std::string> kvs;
+
+        // generate sort key and value
+        for (int j = 0; j < FLAGS_multi_count; j++) {
+            sort_key = generate_string(FLAGS_sortkey_size);
+            value = generate_string(FLAGS_value_size);
+            kvs.emplace(sort_key, value);
+        }
+
+        // write to pegasus
+        int try_count = 0;
+        while (true) {
+            try_count++;
+            int ret = _client->multi_set(hash_key, kvs, FLAGS_pegasus_timeout_ms);
+            if (ret == ::pegasus::PERR_OK) {
+                bytes += (FLAGS_value_size + FLAGS_hashkey_size + FLAGS_sortkey_size) * FLAGS_multi_count;
+                break;
+            } else if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Set returned an error: {}\n", _client->get_error_string(ret));
+                dsn_exit(1);
+            } else {
+                fmt::print(stderr, "Set timeout, retry({})\n", try_count);
+            }
+        }
+
+        // count this operation
+        thread->stats.finished_ops(1, kMultiSet);
+    }
+
+    // count total write bytes
+    thread->stats.add_bytes(bytes);
+}
+
 void benchmark::read_random(thread_arg *thread)
 {
     uint64_t bytes = 0;
@@ -191,6 +241,65 @@ void benchmark::read_random(thread_arg *thread)
     thread->stats.add_message(msg);
 }
 
+void benchmark::multi_get_random(thread_arg *thread)
+{
+    if (FLAGS_benchmark_num % FLAGS_multi_count != 0) {
+        fmt::print(stderr, "num {} should be a multiple of multi_count({}).\n", FLAGS_benchmark_num, FLAGS_multi_count);
+        dsn_exit(1);
+    }
+
+    uint64_t bytes = 0;
+    uint64_t found = 0;
+    int max_fetch_count = 100;
+    int max_fetch_size = 1000000;
+
+    for (int i = 0; i < FLAGS_benchmark_num / FLAGS_multi_count; i++) {
+        // generate hash key
+        std::string hashkey, value;
+        hashkey = generate_string(FLAGS_hashkey_size);
+        std::map<std::string, std::string> kvs;
+        std::set<std::string> sortkeys;
+
+        // generate sort key
+        // generate value for random to keep in peace with write
+        for (int j = 0; j < FLAGS_multi_count; j++) {
+            sortkeys.insert(generate_string(FLAGS_sortkey_size));
+            value = generate_string(FLAGS_value_size);
+        }
+
+        // read from pegasus
+        int try_count = 0;
+        while (true) {
+            try_count++;
+            int ret = _client->multi_get(
+                hashkey, sortkeys, kvs, max_fetch_count, max_fetch_size, FLAGS_pegasus_timeout_ms);
+            if (ret == ::pegasus::PERR_OK) {
+                found += FLAGS_multi_count;
+                bytes += FLAGS_multi_count * hashkey.size();
+                for (auto &kv : kvs) {
+                    bytes = kv.first.size() + kv.second.size() + bytes;
+                }
+                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));
+                dsn_exit(1);
+            } else {
+                fmt::print(stderr, "Get timeout, retry({})\n", try_count);
+            }
+        }
+
+        // count this operation
+        thread->stats.finished_ops(1, kMultiGet);
+    }
+
+    // count total read bytes and hit rate
+    std::string msg = fmt::format("({} of {} found)", found, FLAGS_benchmark_num);
+    thread->stats.add_bytes(bytes);
+    thread->stats.add_message(msg);
+}
+
 void benchmark::delete_random(thread_arg *thread)
 {
     // do delete operation num times
@@ -235,6 +344,10 @@ operation_type benchmark::get_operation_type(const std::string &name)
         op_type = kRead;
     } else if (name == "deleterandom_pegasus") {
         op_type = kDelete;
+    } else if (name == "multisetrandom_pegasus") {
+        op_type = kMultiSet;
+    } else if (name == "multigetrandom_pegasus") {
+        op_type = kMultiGet;
     } else if (!name.empty()) { // No error message for empty name
         fmt::print(stderr, "unknown benchmark '{}'\n", name);
         dsn_exit(1);

src/test/bench_test/benchmark.h

@@ -64,6 +64,8 @@ class benchmark
     void write_random(thread_arg *thread);
     void read_random(thread_arg *thread);
     void delete_random(thread_arg *thread);
+    void multi_set_random(thread_arg *thread);
+    void multi_get_random(thread_arg *thread);
 
     /**  generate hash/sort key and value */
     void generate_kv_pair(std::string &hashkey, std::string &sortkey, std::string &value);

src/test/bench_test/config.cpp

@@ -33,6 +33,7 @@ DSN_DEFINE_int32(pegasus.benchmark, threads, 1, "Number of concurrent threads to
 DSN_DEFINE_int32(pegasus.benchmark, hashkey_size, 16, "size of each hashkey");
 DSN_DEFINE_int32(pegasus.benchmark, sortkey_size, 16, "size of each sortkey");
 DSN_DEFINE_int32(pegasus.benchmark, value_size, 100, "Size of each value");
+DSN_DEFINE_int32(pegasus.benchmark, multi_count, 100, "Values count of the same hashkey");
 
 config::config()
 {

src/test/bench_test/config.ini

@@ -81,4 +81,5 @@ value_size = @VALUE_SIZE@
 hashkey_size = @HASHKEY_SIZE@
 sortkey_size = @SORTKEY_SIZE@
 benchmark_seed = @SEED@
+multi_count = @MULTI_COUNT@
 

src/test/bench_test/statistics.cpp

@@ -26,7 +26,12 @@
 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"}};
+    {kUnknown, "unKnown"},
+    {kRead, "read"},
+    {kWrite, "write"},
+    {kDelete, "delete"},
+    {kMultiSet, "multiSet"},
+    {kMultiGet, "multiGet"}};
 
 statistics::statistics(std::shared_ptr<rocksdb::Statistics> hist_stats)
 {

src/test/bench_test/utils.h

@@ -26,7 +26,9 @@ enum operation_type
     kUnknown = 0,
     kRead,
     kWrite,
-    kDelete
+    kDelete,
+    kMultiGet,
+    kMultiSet
 };
 } // namespace test
 } // namespace pegasus