feat: thread pool

benchmark/src/Benchmark.cpp

@@ -29,6 +29,7 @@
 #include <chrono>
 #include <filesystem>
 #include <memory>
+#include <type_traits>
 
 using namespace std;
 using namespace AllianceDB;
@@ -49,6 +50,14 @@ DEFINE_uint32(num_workers, 2, "Number of workers");
  */
 int main(int argc, char **argv)
 {
+    string usage = "\n";
+    for (auto i = 0; i < extent<decltype(algo_names)>::value; ++i)
+    {
+        usage += "\t" + to_string(i) + ": ";
+        usage += algo_names[i];
+        usage += "\n";
+    }
+    gflags::SetUsageMessage(usage);
     Param param;
     gflags::ParseCommandLineFlags(&argc, &argv, true);
 
@@ -88,17 +97,8 @@ int main(int argc, char **argv)
         break;
     }
     case AlgoType::HandshakeJoin:
-    {
-        auto engine = make_unique<EagerEngine>(param);
-        engine->Run(ctx);
-        break;
-    }
+    case AlgoType::HandshakeJoinOrigin:
     case AlgoType::SplitJoin:
-    {
-        auto engine = make_unique<EagerEngine>(param);
-        engine->Run(ctx);
-        break;
-    }
     case AlgoType::SplitJoinOrigin:
     {
         auto engine = make_unique<EagerEngine>(param);

include/Common/Context.hpp

@@ -4,6 +4,7 @@
 #include "Common/Param.hpp"
 #include "Common/Result.hpp"
 #include "Common/Stream.hpp"
+#include "Utils/ThreadPool.hpp"
 
 namespace AllianceDB
 {
@@ -12,7 +13,10 @@ struct Context
     const Param &param;
     ResultPtr res;
     StreamPtr sr, ss;
-    Context(const Param &param) : param(param), res(std::make_shared<JoinResult>(param)) {}
+    ThreadPool pool;
+    Context(const Param &param)
+        : param(param), res(std::make_shared<JoinResult>(param)), pool(param.num_workers)
+    {}
 };
 
 }  // namespace AllianceDB

include/Common/Types.hpp

@@ -44,17 +44,19 @@ enum class StreamType
 }; /*!< Type of the stream, default R and S */
 enum class AlgoType
 {
-    Verify          = 0,
-    LWJ             = 1,
-    HandshakeJoin   = 2,
-    SplitJoin       = 3,
-    IBWJ            = 4,
-    HashJoin        = 5,
-    SplitJoinOrigin = 6,
+    Verify              = 0,
+    LWJ                 = 1,
+    HandshakeJoin       = 2,
+    SplitJoin           = 3,
+    IBWJ                = 4,
+    HashJoin            = 5,
+    HandshakeJoinOrigin = 6,
+    SplitJoinOrigin     = 7,
 };
 
-constexpr std::string_view algo_names[32] = {
-    "Verify", "LWJ", "HandshakeJoin", "SplitJoin", "IBWJ", "HashJoin", "SplitJoinOrigin",
+constexpr std::string_view algo_names[] = {
+    "Verify", "LWJ",      "HandshakeJoin",       "SplitJoin",
+    "IBWJ",   "HashJoin", "HandshakeJoinOrigin", "SplitJoinOrigin",
 };
 
 using ThreadPtr = std::shared_ptr<std::thread>; /*!< Type of the thread pointer */

include/Utils/ThreadPool.hpp

@@ -0,0 +1,23 @@
+#ifndef INCLUDE_UTILS_THREADPOOL_HPP_
+#define INCLUDE_UTILS_THREADPOOL_HPP_
+
+#include <boost/asio/post.hpp>
+#include <boost/asio/thread_pool.hpp>
+
+class ThreadPool
+{
+public:
+    ThreadPool(size_t num_threads) : num_threads(num_threads), pool(num_threads) {}
+    template <typename T>
+    void Post(T func)
+    {
+        boost::asio::post(pool, func);
+    }
+    void Wait() { pool.join(); }
+
+private:
+    size_t num_threads;
+    boost::asio::thread_pool pool;
+};
+
+#endif

src/Engine/EagerEngine.cpp

@@ -38,6 +38,10 @@ JoinPtr EagerEngine::New()
     {
         return make_shared<HandshakeJoin>(param, windows.size());
     }
+    case AlgoType::HandshakeJoinOrigin:
+    {
+        return make_shared<HandshakeJoin>(param, windows.size());
+    }
     case AlgoType::SplitJoin:
     {
         return make_shared<SplitJoin>(param, windows.size());
@@ -48,7 +52,7 @@ JoinPtr EagerEngine::New()
     }
     default:
     {
-        FATAL("Unsupported algorithm %d", param.algo);
+        FATAL("Unsupported algorithm %d", static_cast<int>(param.algo));
     }
     }
 }
@@ -61,7 +65,7 @@ void EagerEngine::Run(Context &ctx)
     {
         auto nextS = ss->Next(), nextR = sr->Next();
         // no new joiner
-        if (param.algo == AlgoType::SplitJoinOrigin)
+        if (param.algo == AlgoType::SplitJoinOrigin || param.algo == AlgoType::HandshakeJoinOrigin)
         {
             if (nextR->ts == 0)
             {
@@ -100,4 +104,5 @@ void EagerEngine::Run(Context &ctx)
         windows[i]->Wait();
         LOG("algo[%d/%d] joined", i, windows.size());
     }
+    ctx.pool.Wait();
 }

src/Join/HandshakeJoin.cpp

@@ -64,7 +64,9 @@ void HandshakeJoin::Start(Context &ctx)
 {
     for (auto &w : workers)
     {
-        w->Start(ctx);
+        // w->Start(ctx);
+        auto func = [&w, &ctx]() { w->Run(ctx); };
+        ctx.pool.Post(func);
     }
 }
 
@@ -82,12 +84,12 @@ void HandshakeJoin::Wait()
 
 HandshakeJoin::Worker::Worker(const Param &param)
     : param(param),
-      inputr(param.window * 2),
-      inputs(param.window * 2),
+      inputr(param.num_tuples),
+      inputs(param.num_tuples),
       msgi(1),
       msgo(1),
-      left_recv_queue(std::make_shared<spsc_queue<Msg>>(param.window * 2)),
-      right_recv_queue(std::make_shared<spsc_queue<Msg>>(param.window * 2))
+      left_recv_queue(std::make_shared<spsc_queue<Msg>>(param.num_tuples)),
+      right_recv_queue(std::make_shared<spsc_queue<Msg>>(param.num_tuples))
 {}
 
 void HandshakeJoin::Worker::Run(Context &ctx)

src/Join/SplitJoin.cpp

@@ -118,7 +118,8 @@ void SplitJoin::JoinCore::Run(Context &ctx)
 void SplitJoin::JoinCore::Start(Context &ctx)
 {
     auto func = [this, &ctx]() { this->Run(ctx); };
-    t         = make_shared<thread>(func);
+    // t         = make_shared<thread>(func);
+    ctx.pool.Post(func);
 }
 
 void SplitJoin::JoinCore::Store(TuplePtr tuple)