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

Simplify and parallelize CreateExpand

Browse Source
This commit is contained in:
Tyler Neely 2022-11-29 15:07:59 +00:00
parent f8215306e8
commit 7df3a743b9
1 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/query/v2/request_router.hpp
View File

@ -305,24 +305,20 @@ class RequestRouter : public RequestRouterInterface {
// create requests // create requests
InitializeExecutionState(state, new_edges); InitializeExecutionState(state, new_edges);
std::vector<msgs::CreateExpandResponse> responses;
// begin all requests in parallel
for (auto &request : state.requests) { for (auto &request : state.requests) {
auto &storage_client = GetStorageClientForShard(request.shard); auto &storage_client = GetStorageClientForShard(request.shard);
msgs::WriteRequests req = request.request; msgs::ReadRequests req = request.request;
auto write_response_result = storage_client.SendWriteRequest(std::move(req)); request.async_request_token = storage_client.SendAsyncReadRequest(req);
if (write_response_result.HasError()) {
throw std::runtime_error("CreateVertices request timedout");
}
msgs::WriteResponses response_variant = write_response_result.GetValue();
msgs::CreateExpandResponse mapped_response = std::get<msgs::CreateExpandResponse>(response_variant);
if (mapped_response.error) {
throw std::runtime_error("CreateExpand request did not succeed");
}
responses.push_back(mapped_response);
} }
// We are done with this state
state.requests.clear(); // drive requests to completion
std::vector<msgs::ExpandOneResponse> responses;
do {
DriveReadResponses(state, responses);
} while (!state.requests.empty());
return responses; return responses;
} }