Merge pull request #128 from project-tsurugi/scan_default_parallel

add sql.scan_default_parallel config. param.

examples/service_benchmark/main.cpp

@@ -112,6 +112,7 @@ DEFINE_bool(ddl, false, "issue ddl instead of using built-in table. Required for
 DEFINE_bool(secondary, false, "use secondary index");  //NOLINT
 DEFINE_int64(scan_block_size, 100, "max records processed by scan operator before yielding to other tasks");  //NOLINT
 DEFINE_int64(scan_yield_interval, 1, "max time (ms) processed by scan operator before yielding to other tasks");  //NOLINT
+DEFINE_int64(scan_default_parallel, 1, "max parallel execution count of scan tasks");  //NOLINT
 
 namespace tateyama::service_benchmark {
 

include/jogasaki/configuration.h

@@ -490,6 +490,14 @@ class configuration {
         direct_commit_callback_ = arg;
     }
 
+    [[nodiscard]] std::size_t scan_default_parallel() const noexcept {
+        return scan_default_parallel_;
+    }
+
+    void scan_default_parallel(std::size_t arg) noexcept {
+        scan_default_parallel_ = arg;
+    }
+
     friend inline std::ostream& operator<<(std::ostream& out, configuration const& cfg) {
 
         //NOLINTBEGIN
@@ -546,6 +554,7 @@ class configuration {
         print_non_default(rtx_parallel_scan);
         print_non_default(thousandths_ratio_check_local_first);
         print_non_default(direct_commit_callback);
+        print_non_default(scan_default_parallel);
 
         if(cfg.req_cancel_config()) {
             out << "req_cancel_config:" << *cfg.req_cancel_config() << " "; \
@@ -606,6 +615,7 @@ class configuration {
     bool rtx_parallel_scan_ = false;
     std::size_t thousandths_ratio_check_local_first_ = 100;
     bool direct_commit_callback_ = false;
+    std::size_t scan_default_parallel_ = 1;
 
 };
 

src/jogasaki/api/impl/database.cpp

@@ -180,6 +180,7 @@ void dump_public_configurations(configuration const& cfg) {
     LOGCFG << "(dev_rtx_parallel_scan) " << cfg.rtx_parallel_scan() << " : whether to enable parallel scan for RTX";
     LOGCFG << "(dev_thousandths_ratio_check_local_first) " << cfg.thousandths_ratio_check_local_first() << " : how frequently (represented as count out of 1000 executions) task scheduler checks local task queue first";
     LOGCFG << "(dev_direct_commit_callback) " << cfg.direct_commit_callback() << " : whether to make callback directly from shirakami to client on pre-commit response (only for `available` and `accepted`)";
+    LOGCFG << "(scan_default_parallel) " << cfg.scan_default_parallel() << " : max parallel execution count of scan tasks";
 }
 
 status database::start() {

src/jogasaki/api/resource/bridge.cpp

@@ -260,6 +260,9 @@ bool process_sql_config(std::shared_ptr<jogasaki::configuration>& ret, tateyama:
     if (auto v = jogasaki_config->get<bool>("dev_direct_commit_callback")) {
         ret->direct_commit_callback(v.value());
     }
+    if (auto v = jogasaki_config->get<std::size_t>("scan_default_parallel")) {
+        ret->scan_default_parallel(v.value());
+    }
     return true;
 }