From 9b01461f5310296d7c4a00bfaed93b270d239d34 Mon Sep 17 00:00:00 2001 From: WangZhengru Date: Fri, 3 Mar 2023 18:42:15 +0800 Subject: [PATCH 1/3] feat: print algo id in help info --- benchmark/src/Benchmark.cpp | 9 +++++++++ include/Common/Types.hpp | 4 ++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/benchmark/src/Benchmark.cpp b/benchmark/src/Benchmark.cpp index 9b4296cb..c2e63d53 100644 --- a/benchmark/src/Benchmark.cpp +++ b/benchmark/src/Benchmark.cpp @@ -29,6 +29,7 @@ #include #include #include +#include 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::value; ++i) + { + usage += "\t" + to_string(i) + ": "; + usage += algo_names[i]; + usage += "\n"; + } + gflags::SetUsageMessage(usage); Param param; gflags::ParseCommandLineFlags(&argc, &argv, true); diff --git a/include/Common/Types.hpp b/include/Common/Types.hpp index 55a93c69..3c0b2dcc 100644 --- a/include/Common/Types.hpp +++ b/include/Common/Types.hpp @@ -50,10 +50,10 @@ enum class AlgoType SplitJoin = 3, IBWJ = 4, HashJoin = 5, - SplitJoinOrigin = 6, + SplitJoinOrigin = 6 }; -constexpr std::string_view algo_names[32] = { +constexpr std::string_view algo_names[] = { "Verify", "LWJ", "HandshakeJoin", "SplitJoin", "IBWJ", "HashJoin", "SplitJoinOrigin", }; From f5e6663f7610f466229b480edc14188e8ec94f08 Mon Sep 17 00:00:00 2001 From: WangZhengru Date: Wed, 15 Mar 2023 17:29:31 +0800 Subject: [PATCH 2/3] feat: thread pool --- benchmark/src/Benchmark.cpp | 11 +---------- include/Common/Context.hpp | 6 +++++- include/Common/Types.hpp | 18 ++++++++++-------- include/Utils/ThreadPool.hpp | 23 +++++++++++++++++++++++ src/Engine/EagerEngine.cpp | 7 ++++++- src/Join/HandshakeJoin.cpp | 12 +++++++----- src/Join/SplitJoin.cpp | 3 ++- 7 files changed, 54 insertions(+), 26 deletions(-) create mode 100644 include/Utils/ThreadPool.hpp diff --git a/benchmark/src/Benchmark.cpp b/benchmark/src/Benchmark.cpp index c2e63d53..566e94b1 100644 --- a/benchmark/src/Benchmark.cpp +++ b/benchmark/src/Benchmark.cpp @@ -97,17 +97,8 @@ int main(int argc, char **argv) break; } case AlgoType::HandshakeJoin: - { - auto engine = make_unique(param); - engine->Run(ctx); - break; - } + case AlgoType::HandshakeJoinOrigin: case AlgoType::SplitJoin: - { - auto engine = make_unique(param); - engine->Run(ctx); - break; - } case AlgoType::SplitJoinOrigin: { auto engine = make_unique(param); diff --git a/include/Common/Context.hpp b/include/Common/Context.hpp index d64197b5..1a9308ce 100644 --- a/include/Common/Context.hpp +++ b/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 ¶m; ResultPtr res; StreamPtr sr, ss; - Context(const Param ¶m) : param(param), res(std::make_shared(param)) {} + ThreadPool pool; + Context(const Param ¶m) + : param(param), res(std::make_shared(param)), pool(param.num_workers) + {} }; } // namespace AllianceDB diff --git a/include/Common/Types.hpp b/include/Common/Types.hpp index 3c0b2dcc..9fa20d18 100644 --- a/include/Common/Types.hpp +++ b/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[] = { - "Verify", "LWJ", "HandshakeJoin", "SplitJoin", "IBWJ", "HashJoin", "SplitJoinOrigin", + "Verify", "LWJ", "HandshakeJoin", "SplitJoin", + "IBWJ", "HashJoin", "HandshakeJoinOrigin", "SplitJoinOrigin", }; using ThreadPtr = std::shared_ptr; /*!< Type of the thread pointer */ diff --git a/include/Utils/ThreadPool.hpp b/include/Utils/ThreadPool.hpp new file mode 100644 index 00000000..e71d7507 --- /dev/null +++ b/include/Utils/ThreadPool.hpp @@ -0,0 +1,23 @@ +#ifndef INCLUDE_UTILS_THREADPOOL_HPP_ +#define INCLUDE_UTILS_THREADPOOL_HPP_ + +#include +#include + +class ThreadPool +{ +public: + ThreadPool(size_t num_threads) : num_threads(num_threads), pool(num_threads) {} + template + void Post(T func) + { + boost::asio::post(pool, func); + } + void Wait() { pool.join(); } + +private: + size_t num_threads; + boost::asio::thread_pool pool; +}; + +#endif \ No newline at end of file diff --git a/src/Engine/EagerEngine.cpp b/src/Engine/EagerEngine.cpp index fc363700..3593bc3d 100644 --- a/src/Engine/EagerEngine.cpp +++ b/src/Engine/EagerEngine.cpp @@ -38,6 +38,10 @@ JoinPtr EagerEngine::New() { return make_shared(param, windows.size()); } + case AlgoType::HandshakeJoinOrigin: + { + return make_shared(param, windows.size()); + } case AlgoType::SplitJoin: { return make_shared(param, windows.size()); @@ -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(); } diff --git a/src/Join/HandshakeJoin.cpp b/src/Join/HandshakeJoin.cpp index f777bd56..4aa7126f 100644 --- a/src/Join/HandshakeJoin.cpp +++ b/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 ¶m) : 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>(param.window * 2)), - right_recv_queue(std::make_shared>(param.window * 2)) + left_recv_queue(std::make_shared>(param.num_tuples)), + right_recv_queue(std::make_shared>(param.num_tuples)) {} void HandshakeJoin::Worker::Run(Context &ctx) diff --git a/src/Join/SplitJoin.cpp b/src/Join/SplitJoin.cpp index f2fe1511..baa50482 100644 --- a/src/Join/SplitJoin.cpp +++ b/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(func); + // t = make_shared(func); + ctx.pool.Post(func); } void SplitJoin::JoinCore::Store(TuplePtr tuple) From 0a770e04d8edb251fea396cd776f2c2ce78ded06 Mon Sep 17 00:00:00 2001 From: WangZhengru Date: Wed, 15 Mar 2023 17:43:00 +0800 Subject: [PATCH 3/3] fix: static_cast algo --- src/Engine/EagerEngine.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Engine/EagerEngine.cpp b/src/Engine/EagerEngine.cpp index 3593bc3d..d56e552f 100644 --- a/src/Engine/EagerEngine.cpp +++ b/src/Engine/EagerEngine.cpp @@ -52,7 +52,7 @@ JoinPtr EagerEngine::New() } default: { - FATAL("Unsupported algorithm %d", param.algo); + FATAL("Unsupported algorithm %d", static_cast(param.algo)); } } }