diff --git a/benchmark/src/Benchmark.cpp b/benchmark/src/Benchmark.cpp index 9b4296cb..566e94b1 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); @@ -88,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 55a93c69..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[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; /*!< 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..d56e552f 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()); @@ -48,7 +52,7 @@ JoinPtr EagerEngine::New() } default: { - FATAL("Unsupported algorithm %d", param.algo); + FATAL("Unsupported algorithm %d", static_cast(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(); } 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)