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

@@ -54,15 +54,35 @@ DSN_DEFINE_string(
     "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");
+    "\tdeleterandom_pegasus     -- pegasus delete N keys in random order\n"
+    "\tmultisetrandom_pegasus   --pegasus write N random values with multi_count hash keys list\n"
+    "\tmultigetrandom_pegasus   --pegasus read N random keys with multi_count hash list\n");
+
 DSN_DEFINE_validator(benchmarks,
                      [](const char *value) -> bool { return !dsn::utils::is_empty(value); });
 
-DSN_DECLARE_int32(hashkey_size);
-DSN_DECLARE_int32(pegasus_timeout_ms);
-DSN_DECLARE_int32(sortkey_size);
-DSN_DECLARE_int32(threads);
-DSN_DECLARE_int32(value_size);
+DSN_DEFINE_int32(pegasus.benchmark,
+                 pegasus_timeout_ms,
+                 1000,
+                 "pegasus read/write timeout in milliseconds");
+DSN_DEFINE_int32(pegasus.benchmark, threads, 1, "Number of concurrent threads to run");
+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");
+
+DSN_DEFINE_group_validator(multi_count, [](std::string &message) -> bool {
+    std::string operation_type = FLAGS_benchmarks;
+    if ((operation_type == "multisetrandom_pegasus" ||
+         operation_type == "multigetrandom_pegasus") &&
+        FLAGS_benchmark_num % FLAGS_multi_count != 0) {
+        message = fmt::format("num {} should be a multiple of multi_count({}).",
+                              FLAGS_benchmark_num,
+                              FLAGS_multi_count);
+        return false;
+    }
+    return true;
+});
 
 benchmark::benchmark()
 {
@@ -74,6 +94,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}};
 }
 
@@ -170,6 +192,46 @@ void benchmark::write_random(thread_arg *thread)
     thread->stats.add_bytes(bytes);
 }
 
+void benchmark::multi_set_random(thread_arg *thread)
+{
+    uint64_t bytes = 0;
+
+    for (int i = 0; i < FLAGS_benchmark_num / FLAGS_multi_count; i++) {
+        // Generate hash key.
+        std::string hash_key;
+        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++) {
+            kvs.emplace(generate_string(FLAGS_sortkey_size), generate_string(FLAGS_value_size));
+        }
+
+        // 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;
+            }
+            if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Set returned an error: {}\n", _client->get_error_string(ret));
+                dsn_exit(1);
+            }
+            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;
@@ -209,6 +271,61 @@ void benchmark::read_random(thread_arg *thread)
     thread->stats.add_message(msg);
 }
 
+void benchmark::multi_get_random(thread_arg *thread)
+{
+    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;
+        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));
+            // Make output string be sorted like multi_set_random.
+            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;
+            }
+            if (ret != ::pegasus::PERR_TIMEOUT || try_count > 3) {
+                fmt::print(stderr, "Get returned an error: {}\n", _client->get_error_string(ret));
+                dsn_exit(1);
+            }
+            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
@@ -253,6 +370,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

@@ -25,15 +25,6 @@
 namespace pegasus {
 namespace test {
 
-DSN_DEFINE_int32(pegasus.benchmark,
-                 pegasus_timeout_ms,
-                 1000,
-                 "pegasus read/write timeout in milliseconds");
-DSN_DEFINE_int32(pegasus.benchmark, threads, 1, "Number of concurrent threads to run");
-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");
-
 config::config() { env = rocksdb::Env::Default(); }
 } // namespace test
 } // namespace pegasus

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