feat: add sleep_msec_no_row config value to balance CPU usage and T…

…TFB.
SpringQL · laysakura · Jul 13, 2022 · Jul 13, 2022 · Jul 13, 2022 · Jul 13, 2022
commit a484706edfe0196046282682f7a494f11ed37027
diff --git a/springql-core/src/api/spring_config.rs b/springql-core/src/api/spring_config.rs
@@ -18,6 +18,10 @@ n_generic_worker_threads = 1
 # Setting this to > 1 may improve throughput but lead to out-of-order stream processing.
 n_source_worker_threads = 1
 
+# How long a generic worker or a source worker sleeps if it does not receive any row from the upstream.
+# Small number will improve the initial row's E2E latency but increase the CPU usage.
+sleep_msec_no_row = 100
+
 [memory]
 # How much memory is allowed to be used in SpringQL streaming runtime.
 upper_limit_bytes = 10_000_000
@@ -136,6 +140,7 @@ impl SpringConfig {
 pub struct SpringWorkerConfig {
     pub n_generic_worker_threads: u16,
     pub n_source_worker_threads: u16,
+    pub sleep_msec_no_row: u64,
 }
 
 /// Config related to memory management.

diff --git a/springql-core/src/stream_engine/autonomous_executor/task_executor.rs b/springql-core/src/stream_engine/autonomous_executor/task_executor.rs
@@ -50,14 +50,14 @@ impl TaskExecutor {
             repos: repos.clone(),
 
             _generic_worker_pool: GenericWorkerPool::new(
-                config.worker.n_generic_worker_threads,
+                &config.worker,
                 locks.clone(),
                 event_queues.clone(),
                 coordinators.clone(),
                 repos.clone(),
             ),
             _source_worker_pool: SourceWorkerPool::new(
-                config.worker.n_source_worker_threads,
+                &config.worker,
                 locks,
                 event_queues,
                 coordinators,

diff --git a/springql-core/src/stream_engine/autonomous_executor/task_executor/generic_worker_pool.rs b/springql-core/src/stream_engine/autonomous_executor/task_executor/generic_worker_pool.rs
@@ -4,12 +4,15 @@ mod generic_worker;
 
 use std::{cell::RefCell, sync::Arc};
 
-use crate::stream_engine::autonomous_executor::{
-    args::{Coordinators, EventQueues, Locks},
-    repositories::Repositories,
-    task_executor::{
-        generic_worker_pool::generic_worker::GenericWorker,
-        task_worker_thread_handler::{TaskWorkerId, TaskWorkerThreadArg},
+use crate::{
+    api::SpringWorkerConfig,
+    stream_engine::autonomous_executor::{
+        args::{Coordinators, EventQueues, Locks},
+        repositories::Repositories,
+        task_executor::{
+            generic_worker_pool::generic_worker::GenericWorker,
+            task_worker_thread_handler::{TaskWorkerId, TaskWorkerThreadArg},
+        },
     },
 };
 
@@ -26,18 +29,19 @@ pub struct GenericWorkerPool {
 
 impl GenericWorkerPool {
     pub fn new(
-        n_worker_threads: u16,
+        config: &SpringWorkerConfig,
         locks: Locks,
         event_queues: EventQueues,
         coordinators: Coordinators,
         repos: Arc<Repositories>,
     ) -> Self {
-        let workers = (0..n_worker_threads)
+        let workers = (0..config.n_generic_worker_threads)
             .map(|id| {
                 let arg = TaskWorkerThreadArg::new(
                     TaskWorkerId::new(id as u16),
                     locks.task_executor_lock.clone(),
                     repos.clone(),
+                    config.sleep_msec_no_row,
                 );
                 GenericWorker::new(
                     locks.main_job_lock.clone(),

diff --git a/springql-core/src/stream_engine/autonomous_executor/task_executor/source_worker_pool.rs b/springql-core/src/stream_engine/autonomous_executor/task_executor/source_worker_pool.rs
@@ -4,12 +4,15 @@ mod source_worker;
 
 use std::{cell::RefCell, sync::Arc};
 
-use crate::stream_engine::autonomous_executor::{
-    args::{Coordinators, EventQueues, Locks},
-    repositories::Repositories,
-    task_executor::{
-        source_worker_pool::source_worker::SourceWorker,
-        task_worker_thread_handler::{TaskWorkerId, TaskWorkerThreadArg},
+use crate::{
+    api::SpringWorkerConfig,
+    stream_engine::autonomous_executor::{
+        args::{Coordinators, EventQueues, Locks},
+        repositories::Repositories,
+        task_executor::{
+            source_worker_pool::source_worker::SourceWorker,
+            task_worker_thread_handler::{TaskWorkerId, TaskWorkerThreadArg},
+        },
     },
 };
 
@@ -26,18 +29,19 @@ pub struct SourceWorkerPool {
 
 impl SourceWorkerPool {
     pub fn new(
-        n_worker_threads: u16,
+        config: &SpringWorkerConfig,
         locks: Locks,
         event_queues: EventQueues,
         coordinators: Coordinators,
         repos: Arc<Repositories>,
     ) -> Self {
-        let workers = (0..n_worker_threads)
+        let workers = (0..config.n_source_worker_threads)
             .map(|id| {
                 let arg = TaskWorkerThreadArg::new(
                     TaskWorkerId::new(id as u16),
                     locks.task_executor_lock.clone(),
                     repos.clone(),
+                    config.sleep_msec_no_row,
                 );
                 SourceWorker::new(
                     locks.main_job_lock.clone(),

diff --git a/...ql-core/src/stream_engine/autonomous_executor/task_executor/task_worker_thread_handler.rs b/...ql-core/src/stream_engine/autonomous_executor/task_executor/task_worker_thread_handler.rs
@@ -35,6 +35,7 @@ pub struct TaskWorkerThreadArg {
     pub worker_id: TaskWorkerId,
     task_executor_lock: Arc<TaskExecutorLock>,
     repos: Arc<Repositories>,
+    sleep_msec_no_row: u64,
 }
 
 #[derive(Debug)]
@@ -93,7 +94,7 @@ impl TaskWorkerThreadHandler {
                     );
                     if processed_rows.is_empty() {
                         // Wait for rows to process
-                        thread::sleep(Duration::from_millis(TASK_WAIT_MSEC));
+                        thread::sleep(Duration::from_millis(thread_arg.sleep_msec_no_row));
                     }
                 } else {
                     // Wait for tasks to execute

diff --git a/springql/tests/feat_worker_config.rs b/springql/tests/feat_worker_config.rs
@@ -11,9 +11,15 @@ use springql_foreign_service::{
 };
 use springql_test_logger::setup_test_logger;
 
-fn t(worker_config: SpringWorkerConfig) {
+fn t(n_generic_worker_threads: u16, n_source_worker_threads: u16) {
     setup_test_logger();
 
+    let worker_config = SpringWorkerConfig {
+        n_generic_worker_threads,
+        n_source_worker_threads,
+        sleep_msec_no_row: 100,
+    };
+
     let json_oracle = json!({
         "ts": "2021-11-04 23:02:52.123456789",
         "ticker": "ORCL",
@@ -100,32 +106,20 @@ fn t(worker_config: SpringWorkerConfig) {
 
 #[test]
 fn test_feat_1generic_1source() {
-    t(SpringWorkerConfig {
-        n_generic_worker_threads: 1,
-        n_source_worker_threads: 1,
-    })
+    t(1, 1)
 }
 
 #[test]
 fn test_feat_5generic_1source() {
-    t(SpringWorkerConfig {
-        n_generic_worker_threads: 5,
-        n_source_worker_threads: 1,
-    })
+    t(5, 1)
 }
 
 #[test]
 fn test_feat_1generic_5source() {
-    t(SpringWorkerConfig {
-        n_generic_worker_threads: 1,
-        n_source_worker_threads: 5,
-    })
+    t(1, 5)
 }
 
 #[test]
 fn test_feat_5generic_5source() {
-    t(SpringWorkerConfig {
-        n_generic_worker_threads: 5,
-        n_source_worker_threads: 5,
-    })
+    t(5, 5)
 }