make the threadpool parametrized from settings.h

Author: pbollimp

src/common/init.cpp

@@ -40,11 +40,11 @@ void PelotonInit::Initialize() {
   thread_pool.Initialize(0, std::thread::hardware_concurrency() + 3);
 
   // start worker pool
-  threadpool::MonoQueuePool::GetInstance(32, 4).Startup();
+  threadpool::MonoQueuePool::GetInstance().Startup();
 
   // start brain thread pool
   if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
-    threadpool::MonoQueuePool::GetBrainInstance(32, 1).Startup();
+    threadpool::MonoQueuePool::GetBrainInstance().Startup();
   }
 
   int parallelism = (std::thread::hardware_concurrency() + 3) / 4;
@@ -110,11 +110,11 @@ void PelotonInit::Shutdown() {
   concurrency::EpochManagerFactory::GetInstance().StopEpoch();
 
   // stop worker pool
-  threadpool::MonoQueuePool::GetInstance(32, 4).Shutdown();
+  threadpool::MonoQueuePool::GetInstance().Shutdown();
 
   // stop brain thread pool
   if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
-    threadpool::MonoQueuePool::GetBrainInstance(32, 1).Shutdown();
+    threadpool::MonoQueuePool::GetBrainInstance().Shutdown();
   }
 
   thread_pool.Shutdown();

src/gc/transaction_level_gc_manager.cpp

@@ -136,7 +136,7 @@ int TransactionLevelGCManager::Unlink(const int &thread_id,
       std::vector<std::string> query_strings = txn_ctx->GetQueryStrings();
       if(query_strings.size() != 0) {
         uint64_t timestamp = txn_ctx->GetTimestamp();
-        auto &pool = threadpool::MonoQueuePool::GetBrainInstance(32, 1);
+        auto &pool = threadpool::MonoQueuePool::GetBrainInstance();
         for(auto query_string: query_strings) {
           pool.SubmitTask([this, query_string, timestamp] {
             brain::QueryLogger::LogQuery(query_string, timestamp);

src/include/settings/settings.h

@@ -71,6 +71,18 @@ SETTING_double(bnlj_buffer_size,
              1.0 * 1024.0 * 1024.0,
              true, true)
 
+// Size of the MonoQueue task queue
+SETTING_int(monoqueue_task_queue_size,
+            "MonoQueue Task Queue Size (default: 32)",
+            32,
+            false, false)
+
+// Size of the MonoQueue worker pool
+SETTING_int(monoqueue_worker_pool_size,
+            "MonoQueue Worker Pool Size (default: 4)",
+            4,
+            false, false)
+
 //===----------------------------------------------------------------------===//
 // WRITE AHEAD LOG
 //===----------------------------------------------------------------------===//
@@ -125,6 +137,17 @@ SETTING_bool(brain,
             false,
             true, true)
 
+// Size of the brain task queue
+SETTING_int(brain_task_queue_size,
+            "Brain Task Queue Size (default: 32)",
+            32,
+            false, false)
+
+// Size of the brain worker pool
+SETTING_int(brain_worker_pool_size,
+            "Brain Worker Pool Size (default: 1)",
+            1,
+            false, false)
 
 //===----------------------------------------------------------------------===//
 // CODEGEN

src/include/threadpool/mono_queue_pool.h

@@ -12,6 +12,7 @@
 
 #pragma once
 
+#include "settings/settings_manager.h"
 #include "worker_pool.h"
 
 namespace peloton {
@@ -51,13 +52,21 @@ class MonoQueuePool {
     task_queue_.Enqueue(std::move(func));
   }
 
-  static MonoQueuePool &GetInstance(int task_queue_size, int worker_pool_size) {
+  static MonoQueuePool &GetInstance() {
+    int task_queue_size = settings::SettingsManager::GetBool(
+        settings::SettingId::monoqueue_task_queue_size);
+    int worker_pool_size = settings::SettingsManager::GetBool(
+        settings::SettingId::monoqueue_worker_pool_size);
+
     static MonoQueuePool mono_queue_pool(task_queue_size, worker_pool_size);
     return mono_queue_pool;
   }
 
-  static MonoQueuePool &GetBrainInstance(int task_queue_size,
-                                         int worker_pool_size) {
+  static MonoQueuePool &GetBrainInstance() {
+    int task_queue_size = settings::SettingsManager::GetBool(
+        settings::SettingId::brain_task_queue_size);
+    int worker_pool_size = settings::SettingsManager::GetBool(
+        settings::SettingId::brain_worker_pool_size);
     static MonoQueuePool brain_queue_pool(task_queue_size, worker_pool_size);
     return brain_queue_pool;
   }

src/traffic_cop/traffic_cop.cpp

@@ -187,7 +187,7 @@ executor::ExecutionResult TrafficCop::ExecuteHelper(
     task_callback_(task_callback_arg_);
   };
 
-  auto &pool = threadpool::MonoQueuePool::GetInstance(32, 4);
+  auto &pool = threadpool::MonoQueuePool::GetInstance();
   pool.SubmitTask([plan, txn, &params, &result, &result_format, on_complete] {
     executor::PlanExecutor::ExecutePlan(plan, txn, params, result_format,
                                         on_complete);