tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add test for PlanDispatcher and PlanConsumer

Browse Source 
Summary:
Added test for `PlanDispatcher` and `PlanConsumer`.
This diff also contains a fix for the async rpc call on all clients.h

Reviewers: florijan, teon.banek

Reviewed By: florijan

Subscribers: pullbot, dgleich

Differential Revision: https://phabricator.memgraph.io/D1135
This commit is contained in:
Matija Santl 2018-01-24 09:54:38 +01:00
parent 62323965e3
commit 31aded2dae
3 changed files with 37 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/distributed/plan_dispatcher.cpp
View File

@ -15,6 +15,10 @@ void PlanDispatcher::DispatchPlan(
client.Call<DistributedPlanRpc>(300ms, plan_id, plan, symbol_table); client.Call<DistributedPlanRpc>(300ms, plan_id, plan, symbol_table);
CHECK(result) << "Failed to dispatch plan to worker"; CHECK(result) << "Failed to dispatch plan to worker";
}); });
for (auto &future : futures) {
future.wait();
}
} }
} // namespace distributed } // namespace distributed

6
src/distributed/rpc_worker_clients.hpp
View File

@ -58,9 +58,9 @@ class RpcWorkerClients {
if (worker_id == skip_worker_id) continue; if (worker_id == skip_worker_id) continue;
auto &client = GetClient(worker_id); auto &client = GetClient(worker_id);
futures.emplace_back( futures.emplace_back(std::async(std::launch::async, [execute, &client]() {
std::async(std::launch::async, return execute(client);
[&execute, &client]() { return execute(client); })); }));
} }
return futures; return futures;
} }

30
tests/unit/distributed_graph_db.cpp
View File

@ -8,9 +8,12 @@
#include "distributed/coordination.hpp" #include "distributed/coordination.hpp"
#include "distributed/coordination_master.hpp" #include "distributed/coordination_master.hpp"
#include "distributed/coordination_worker.hpp" #include "distributed/coordination_worker.hpp"
#include "distributed/plan_consumer.hpp"
#include "distributed/plan_dispatcher.hpp"
#include "distributed/remote_data_rpc_clients.hpp" #include "distributed/remote_data_rpc_clients.hpp"
#include "distributed/remote_data_rpc_server.hpp" #include "distributed/remote_data_rpc_server.hpp"
#include "io/network/endpoint.hpp" #include "io/network/endpoint.hpp"
#include "query_plan_common.hpp"
#include "transactions/engine_master.hpp" #include "transactions/engine_master.hpp"
template <typename T> template <typename T>
@ -201,3 +204,30 @@ TEST_F(DistributedGraphDbTest, RemoteDataGetting) {
EXPECT_EQ(e1_in_w2.PropsAt(w2_dba.Property("p3")).Value<bool>(), true); EXPECT_EQ(e1_in_w2.PropsAt(w2_dba.Property("p3")).Value<bool>(), true);
} }
} }
TEST_F(DistributedGraphDbTest, DispatchPlan) {
auto kRPCWaitTime = 600ms;
int64_t plan_id = 5;
SymbolTable symbol_table;
AstTreeStorage storage;
auto scan_all = MakeScanAll(storage, symbol_table, "n");
master().plan_dispatcher().DispatchPlan(plan_id, scan_all.op_, symbol_table);
std::this_thread::sleep_for(kRPCWaitTime);
{
auto cached = worker1().plan_consumer().PlanForId(plan_id);
EXPECT_NE(dynamic_cast<query::plan::ScanAll *>(cached.first.get()),
nullptr);
EXPECT_EQ(cached.second.max_position(), symbol_table.max_position());
EXPECT_EQ(cached.second.table(), symbol_table.table());
}
{
auto cached = worker2().plan_consumer().PlanForId(plan_id);
EXPECT_NE(dynamic_cast<query::plan::ScanAll *>(cached.first.get()),
nullptr);
EXPECT_EQ(cached.second.max_position(), symbol_table.max_position());
EXPECT_EQ(cached.second.table(), symbol_table.table());
}
}