add config parameter dev_thousandths_ratio_check_local_first

project-tsurugi/tsurugi-issues#706

include/jogasaki/configuration.h

@@ -474,6 +474,14 @@ class configuration {
         rtx_parallel_scan_ = arg;
     }
 
+    [[nodiscard]] std::size_t thousandths_ratio_check_local_first() const noexcept {
+        return thousandths_ratio_check_local_first_;
+    }
+
+    void thousandths_ratio_check_local_first(std::size_t arg) noexcept {
+        thousandths_ratio_check_local_first_ = arg;
+    }
+
     friend inline std::ostream& operator<<(std::ostream& out, configuration const& cfg) {
 
         //NOLINTBEGIN
@@ -528,6 +536,7 @@ class configuration {
         print_non_default(scan_block_size);
         print_non_default(scan_yield_interval);
         print_non_default(rtx_parallel_scan);
+        print_non_default(thousandths_ratio_check_local_first);
 
         if(cfg.req_cancel_config()) {
             out << "req_cancel_config:" << *cfg.req_cancel_config() << " "; \
@@ -586,6 +595,7 @@ class configuration {
     std::size_t scan_block_size_ = 100;
     std::size_t scan_yield_interval_ = 1;
     bool rtx_parallel_scan_ = false;
+    std::size_t thousandths_ratio_check_local_first_ = 100;
 
 };
 

src/jogasaki/api/impl/database.cpp

@@ -178,6 +178,7 @@ void dump_public_configurations(configuration const& cfg) {
     LOGCFG << "(scan_block_size) " << cfg.scan_block_size() << " : max records processed by scan operator before yielding to other task";
     LOGCFG << "(scan_yield_interval) " << cfg.scan_yield_interval() << " : max time (ms) processed by scan operator before yielding to other tasks";
     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";
 }
 
 status database::start() {

src/jogasaki/api/resource/bridge.cpp

@@ -254,6 +254,9 @@ bool process_sql_config(std::shared_ptr<jogasaki::configuration>& ret, tateyama:
     if (auto v = jogasaki_config->get<bool>("dev_rtx_parallel_scan")) {
         ret->rtx_parallel_scan(v.value());
     }
+    if (auto v = jogasaki_config->get<std::size_t>("dev_thousandths_ratio_check_local_first")) {
+        ret->thousandths_ratio_check_local_first(v.value());
+    }
     return true;
 }
 

src/jogasaki/scheduler/stealing_task_scheduler.cpp

@@ -159,6 +159,7 @@ tateyama::task_scheduler::task_scheduler_cfg stealing_task_scheduler::create_sch
     ret.watcher_interval(params.watcher_interval());
     ret.worker_try_count(params.worker_try_count());
     ret.worker_suspend_timeout(params.worker_suspend_timeout());
+    ret.ratio_check_local_first({params.thousandths_ratio_check_local_first(), 1000});
     return ret;
 }
 

src/jogasaki/scheduler/thread_params.h

@@ -38,7 +38,8 @@ class thread_params {
         bool busy_worker,
         std::size_t watcher_interval,
         std::size_t worker_try_count,
-        std::size_t worker_suspend_timeout
+        std::size_t worker_suspend_timeout,
+        std::size_t thousandths_ratio_check_local_first
     ) :
         threads_(threads),
         set_core_affinity_(set_core_affinity),
@@ -53,7 +54,8 @@ class thread_params {
         busy_worker_(busy_worker),
         watcher_interval_(watcher_interval),
         worker_try_count_(worker_try_count),
-        worker_suspend_timeout_(worker_suspend_timeout)
+        worker_suspend_timeout_(worker_suspend_timeout),
+        thousandths_ratio_check_local_first_(thousandths_ratio_check_local_first)
     {}
 
     explicit thread_params(std::shared_ptr<configuration> const& cfg) :
@@ -71,7 +73,8 @@ class thread_params {
             cfg->busy_worker(),
             cfg->watcher_interval(),
             cfg->worker_try_count(),
-            cfg->worker_suspend_timeout()
+            cfg->worker_suspend_timeout(),
+            cfg->thousandths_ratio_check_local_first()
         )
     {}
 
@@ -147,6 +150,13 @@ class thread_params {
         worker_suspend_timeout_ = arg;
     }
 
+    [[nodiscard]] std::size_t thousandths_ratio_check_local_first() const noexcept {
+        return thousandths_ratio_check_local_first_;
+    }
+
+    void thousandths_ratio_check_local_first(std::size_t arg) noexcept {
+        thousandths_ratio_check_local_first_ = arg;
+    }
 private:
     std::size_t threads_{};
     bool set_core_affinity_{};
@@ -162,6 +172,7 @@ class thread_params {
     std::size_t watcher_interval_{};
     std::size_t worker_try_count_{};
     std::size_t worker_suspend_timeout_{};
+    std::size_t thousandths_ratio_check_local_first_{};
 };
 
 } // namespace