merge the two QueuePools together

Author: pbollimp

src/common/init.cpp

@@ -24,7 +24,6 @@
 #include "gc/gc_manager_factory.h"
 #include "settings/settings_manager.h"
 #include "threadpool/mono_queue_pool.h"
-#include "threadpool/brain_thread_pool.h"
 
 namespace peloton {
 
@@ -41,11 +40,11 @@ void PelotonInit::Initialize() {
   thread_pool.Initialize(0, std::thread::hardware_concurrency() + 3);
 
   // start worker pool
-  threadpool::MonoQueuePool::GetInstance().Startup();
+  threadpool::MonoQueuePool::GetInstance(32, 4).Startup();
 
   // start brain thread pool
   if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
-    threadpool::BrainThreadPool::GetInstance().Startup();
+    threadpool::MonoQueuePool::GetBrainInstance(32, 1).Startup();
   }
 
   int parallelism = (std::thread::hardware_concurrency() + 3) / 4;
@@ -111,11 +110,11 @@ void PelotonInit::Shutdown() {
   concurrency::EpochManagerFactory::GetInstance().StopEpoch();
 
   // stop worker pool
-  threadpool::MonoQueuePool::GetInstance().Shutdown();
+  threadpool::MonoQueuePool::GetInstance(32, 4).Shutdown();
 
   // stop brain thread pool
   if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
-    threadpool::BrainThreadPool::GetInstance().Shutdown();
+    threadpool::MonoQueuePool::GetBrainInstance(32, 1).Shutdown();
   }
 
   thread_pool.Shutdown();

src/gc/transaction_level_gc_manager.cpp

@@ -22,7 +22,7 @@
 #include "storage/tile_group.h"
 #include "storage/tuple.h"
 #include "storage/storage_manager.h"
-#include "threadpool/brain_thread_pool.h"
+#include "threadpool/mono_queue_pool.h"
 
 
 namespace peloton {
@@ -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::BrainThreadPool::GetInstance();
+        auto &pool = threadpool::MonoQueuePool::GetBrainInstance(32, 1);
         for(auto query_string: query_strings) {
           pool.SubmitTask([this, query_string, timestamp] {
             brain::QueryLogger::LogQuery(query_string, timestamp);

src/include/threadpool/brain_thread_pool.h

@@ -17,19 +17,15 @@
 namespace peloton {
 namespace threadpool {
 
-// TODO: tune these variables
-constexpr static size_t kDefaultTaskQueueSize = 32;
-constexpr static size_t kDefaultWorkerPoolSize = 4;
-
 /**
  * @brief Wrapper class for single queue and single pool.
  * One should use this if possible.
  */
 class MonoQueuePool {
  public:
-  MonoQueuePool()
-      : task_queue_(kDefaultTaskQueueSize),
-        worker_pool_(kDefaultWorkerPoolSize, &task_queue_),
+  MonoQueuePool(int task_queue_size = 32, int worker_pool_size = 4)
+      : task_queue_(task_queue_size),
+        worker_pool_(worker_pool_size, &task_queue_),
         is_running_(false) {}
 
   ~MonoQueuePool() {
@@ -55,11 +51,17 @@ class MonoQueuePool {
     task_queue_.Enqueue(std::move(func));
   }
 
-  static MonoQueuePool &GetInstance() {
-    static MonoQueuePool mono_queue_pool;
+  static MonoQueuePool &GetInstance(int task_queue_size, int 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 brain_queue_pool(task_queue_size, worker_pool_size);
+    return brain_queue_pool;
+  }
+
  private:
   TaskQueue task_queue_;
   WorkerPool worker_pool_;

src/include/threadpool/mono_queue_pool.h

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