Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor storage client
Browse files Browse the repository at this point in the history
yixinglu committed Mar 16, 2022
1 parent a57acd0 commit 5814d59
Showing 11 changed files with 278 additions and 357 deletions.
4 changes: 2 additions & 2 deletions .github/workflows/pull_request.yml
Original file line number Diff line number Diff line change
@@ -159,7 +159,7 @@ jobs:
;;
esac
working-directory: tests/
timeout-minutes: 2
timeout-minutes: 4
- name: Pytest
run: |
make RM_DIR=false DEBUG=false J=${{ steps.cmake.outputs.j }} test
@@ -266,7 +266,7 @@ jobs:
run: |
make standalone-up
working-directory: tests/
timeout-minutes: 60
timeout-minutes: 4
- name: TCK
run: |
make RM_DIR=false DEBUG=false J=${{ steps.cmake.outputs.j }} standalone-tck
15 changes: 9 additions & 6 deletions src/clients/storage/InternalStorageClient.cpp
Original file line number Diff line number Diff line change
@@ -64,9 +64,10 @@ void InternalStorageClient::chainUpdateEdge(cpp2::UpdateEdgeRequest& reversedReq
}
auto resp = getResponse(
evb,
std::make_pair(leader, chainReq),
leader,
chainReq,
[](cpp2::InternalStorageServiceAsyncClient* client, const cpp2::ChainUpdateEdgeRequest& r) {
return client->future_chainUpdateEdge(r);
return client->semifuture_chainUpdateEdge(r);
});

std::move(resp).thenTry([=, p = std::move(p)](auto&& t) mutable {
@@ -102,9 +103,10 @@ void InternalStorageClient::chainAddEdges(cpp2::AddEdgesRequest& directReq,
cpp2::ChainAddEdgesRequest chainReq = makeChainAddReq(directReq, termId, optVersion);
auto resp = getResponse(
evb,
std::make_pair(leader, chainReq),
leader,
chainReq,
[](cpp2::InternalStorageServiceAsyncClient* client, const cpp2::ChainAddEdgesRequest& r) {
return client->future_chainAddEdges(r);
return client->semifuture_chainAddEdges(r);
});

std::move(resp).thenTry([=, p = std::move(p)](auto&& t) mutable {
@@ -158,9 +160,10 @@ void InternalStorageClient::chainDeleteEdges(cpp2::DeleteEdgesRequest& req,
chainReq.term_ref() = termId;
auto resp = getResponse(
evb,
std::make_pair(leader, chainReq),
leader,
chainReq,
[](cpp2::InternalStorageServiceAsyncClient* client, const cpp2::ChainDeleteEdgesRequest& r) {
return client->future_chainDeleteEdges(r);
return client->semifuture_chainDeleteEdges(r);
});

std::move(resp).thenTry([=, p = std::move(p)](auto&& t) mutable {
59 changes: 25 additions & 34 deletions src/clients/storage/StorageClient.cpp
Original file line number Diff line number Diff line change
@@ -103,7 +103,7 @@ StorageRpcRespFuture<cpp2::GetNeighborsResponse> StorageClient::getNeighbors(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::GetNeighborsRequest& r) {
return client->future_getNeighbors(r);
return client->semifuture_getNeighbors(r);
});
}

@@ -142,7 +142,7 @@ StorageRpcRespFuture<cpp2::ExecResponse> StorageClient::addVertices(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::AddVerticesRequest& r) {
return client->future_addVertices(r);
return client->semifuture_addVertices(r);
});
}

@@ -180,8 +180,8 @@ StorageRpcRespFuture<cpp2::ExecResponse> StorageClient::addEdges(const CommonReq
std::move(requests),
[useToss = param.useExperimentalFeature](ThriftClientType* client,
const cpp2::AddEdgesRequest& r) {
return useToss ? client->future_chainAddEdges(r)
: client->future_addEdges(r);
return useToss ? client->semifuture_chainAddEdges(r)
: client->semifuture_addEdges(r);
});
}

@@ -237,7 +237,7 @@ StorageRpcRespFuture<cpp2::GetPropResponse> StorageClient::getProps(

return collectResponse(
param.evb, std::move(requests), [](ThriftClientType* client, const cpp2::GetPropRequest& r) {
return client->future_getProps(r);
return client->semifuture_getProps(r);
});
}

@@ -270,8 +270,8 @@ StorageRpcRespFuture<cpp2::ExecResponse> StorageClient::deleteEdges(
std::move(requests),
[useToss = param.useExperimentalFeature](
ThriftClientType* client, const cpp2::DeleteEdgesRequest& r) {
return useToss ? client->future_chainDeleteEdges(r)
: client->future_deleteEdges(r);
return useToss ? client->semifuture_chainDeleteEdges(r)
: client->semifuture_deleteEdges(r);
});
}

@@ -303,7 +303,7 @@ StorageRpcRespFuture<cpp2::ExecResponse> StorageClient::deleteVertices(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::DeleteVerticesRequest& r) {
return client->future_deleteVertices(r);
return client->semifuture_deleteVertices(r);
});
}

@@ -335,7 +335,7 @@ StorageRpcRespFuture<cpp2::ExecResponse> StorageClient::deleteTags(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::DeleteTagsRequest& r) {
return client->future_deleteTags(r);
return client->semifuture_deleteTags(r);
});
}

@@ -352,8 +352,6 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateVert
return folly::makeFuture<StatusOr<storage::cpp2::UpdateResponse>>(cbStatus.status());
}

std::pair<HostAddr, cpp2::UpdateVertexRequest> request;

DCHECK(!!metaClient_);
auto status = metaClient_->partsNum(param.space);
if (!status.ok()) {
@@ -370,7 +368,6 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateVert
if (!host.ok()) {
return folly::makeFuture<StatusOr<storage::cpp2::UpdateResponse>>(host.status());
}
request.first = std::move(host).value();
cpp2::UpdateVertexRequest req;
req.space_id_ref() = param.space;
req.vertex_id_ref() = vertexId;
@@ -383,12 +380,12 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateVert
if (condition.size() > 0) {
req.condition_ref() = std::move(condition);
}
request.second = std::move(req);

return getResponse(param.evb,
std::move(request),
host.value(),
req,
[](ThriftClientType* client, const cpp2::UpdateVertexRequest& r) {
return client->future_updateVertex(r);
return client->semifuture_updateVertex(r);
});
}

@@ -405,8 +402,6 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateEdge
return folly::makeFuture<StatusOr<storage::cpp2::UpdateResponse>>(cbStatus.status());
}

std::pair<HostAddr, cpp2::UpdateEdgeRequest> request;

DCHECK(!!metaClient_);
auto status = metaClient_->partsNum(space);
if (!status.ok()) {
@@ -423,7 +418,6 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateEdge
if (!host.ok()) {
return folly::makeFuture<StatusOr<storage::cpp2::UpdateResponse>>(host.status());
}
request.first = std::move(host).value();
cpp2::UpdateEdgeRequest req;
req.space_id_ref() = space;
req.edge_key_ref() = edgeKey;
@@ -435,21 +429,20 @@ folly::Future<StatusOr<storage::cpp2::UpdateResponse>> StorageClient::updateEdge
if (condition.size() > 0) {
req.condition_ref() = std::move(condition);
}
request.second = std::move(req);

return getResponse(param.evb,
std::move(request),
host.value(),
req,
[useExperimentalFeature = param.useExperimentalFeature](
ThriftClientType* client, const cpp2::UpdateEdgeRequest& r) {
return useExperimentalFeature ? client->future_chainUpdateEdge(r)
: client->future_updateEdge(r);
return useExperimentalFeature ? client->semifuture_chainUpdateEdge(r)
: client->semifuture_updateEdge(r);
});
}

folly::Future<StatusOr<cpp2::GetUUIDResp>> StorageClient::getUUID(GraphSpaceID space,
const std::string& name,
folly::EventBase* evb) {
std::pair<HostAddr, cpp2::GetUUIDReq> request;
DCHECK(!!metaClient_);
auto status = metaClient_->partsNum(space);
if (!status.ok()) {
@@ -466,16 +459,14 @@ folly::Future<StatusOr<cpp2::GetUUIDResp>> StorageClient::getUUID(GraphSpaceID s
if (!host.ok()) {
return folly::makeFuture<StatusOr<storage::cpp2::GetUUIDResp>>(host.status());
}
request.first = std::move(host).value();
cpp2::GetUUIDReq req;
req.space_id_ref() = space;
req.part_id_ref() = part;
req.name_ref() = name;
request.second = std::move(req);

return getResponse(
evb, std::move(request), [](ThriftClientType* client, const cpp2::GetUUIDReq& r) {
return client->future_getUUID(r);
evb, host.value(), req, [](ThriftClientType* client, const cpp2::GetUUIDReq& r) {
return client->semifuture_getUUID(r);
});
}

@@ -523,7 +514,7 @@ StorageRpcRespFuture<cpp2::LookupIndexResp> StorageClient::lookupIndex(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::LookupIndexRequest& r) {
return client->future_lookupIndex(r);
return client->semifuture_lookupIndex(r);
});
}

@@ -552,7 +543,7 @@ StorageRpcRespFuture<cpp2::GetNeighborsResponse> StorageClient::lookupAndTravers
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::LookupAndTraverseRequest& r) {
return client->future_lookupAndTraverse(r);
return client->semifuture_lookupAndTraverse(r);
});
}

@@ -583,7 +574,7 @@ StorageRpcRespFuture<cpp2::ScanResponse> StorageClient::scanEdge(

return collectResponse(
param.evb, std::move(requests), [](ThriftClientType* client, const cpp2::ScanEdgeRequest& r) {
return client->future_scanEdge(r);
return client->semifuture_scanEdge(r);
});
}

@@ -615,7 +606,7 @@ StorageRpcRespFuture<cpp2::ScanResponse> StorageClient::scanVertex(
return collectResponse(param.evb,
std::move(requests),
[](ThriftClientType* client, const cpp2::ScanVertexRequest& r) {
return client->future_scanVertex(r);
return client->semifuture_scanVertex(r);
});
}

@@ -641,7 +632,7 @@ folly::SemiFuture<StorageRpcResponse<cpp2::KVGetResponse>> StorageClient::get(

return collectResponse(
evb, std::move(requests), [](ThriftClientType* client, const cpp2::KVGetRequest& r) {
return client->future_get(r);
return client->semifuture_get(r);
});
}

@@ -666,7 +657,7 @@ folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> StorageClient::put(

return collectResponse(
evb, std::move(requests), [](ThriftClientType* client, const cpp2::KVPutRequest& r) {
return client->future_put(r);
return client->semifuture_put(r);
});
}

@@ -691,7 +682,7 @@ folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>> StorageClient::remove(

return collectResponse(
evb, std::move(requests), [](ThriftClientType* client, const cpp2::KVRemoveRequest& r) {
return client->future_remove(r);
return client->semifuture_remove(r);
});
}

339 changes: 124 additions & 215 deletions src/clients/storage/StorageClientBase-inl.h

Large diffs are not rendered by default.

13 changes: 3 additions & 10 deletions src/clients/storage/StorageClientBase.h
Original file line number Diff line number Diff line change
@@ -12,6 +12,7 @@
#include "clients/meta/MetaClient.h"
#include "common/base/Base.h"
#include "common/base/StatusOr.h"
#include "common/datatypes/HostAddr.h"
#include "common/meta/Common.h"
#include "common/thrift/ThriftClientManager.h"
#include "interface/gen-cpp2/storage_types.h"
@@ -144,18 +145,10 @@ class StorageClientBase {
class Response = typename std::result_of<RemoteFunc(ClientType* client,
const Request&)>::type::value_type>
folly::Future<StatusOr<Response>> getResponse(folly::EventBase* evb,
std::pair<HostAddr, Request>&& request,
const HostAddr& host,
const Request& request,
RemoteFunc&& remoteFunc);

template <class Request,
class RemoteFunc,
class Response = typename std::result_of<RemoteFunc(ClientType* client,
const Request&)>::type::value_type>
void getResponseImpl(folly::EventBase* evb,
std::pair<HostAddr, Request> request,
RemoteFunc remoteFunc,
std::shared_ptr<folly::Promise<StatusOr<Response>>> pro);

// Cluster given ids into the host they belong to
// The method returns a map
// host_addr (A host, but in most case, the leader will be chosen)
17 changes: 17 additions & 0 deletions src/common/base/StatusOr.h
Original file line number Diff line number Diff line change
@@ -6,6 +6,8 @@
#ifndef COMMON_BASE_STATUSOR_H_
#define COMMON_BASE_STATUSOR_H_

#include <type_traits>

#include "common/base/Base.h"
#include "common/base/Status.h"

@@ -336,6 +338,21 @@ class StatusOr final {
uint8_t state_;
};

namespace internal {
template <typename T>
struct StatusOrValueType {
using type = T;
};

template <typename T>
struct StatusOrValueType<StatusOr<T>> {
using type = std::remove_cv_t<std::remove_reference_t<T>>;
};
} // namespace internal

template <typename T>
using status_or_value_t = typename internal::StatusOrValueType<T>::type;

} // namespace nebula

#endif // COMMON_BASE_STATUSOR_H_
92 changes: 46 additions & 46 deletions src/graph/executor/test/StorageServerStub.cpp
Original file line number Diff line number Diff line change
@@ -8,12 +8,12 @@
#include "common/base/Base.h"
#include "storage/GraphStorageLocalServer.h"

#define LOCAL_RETURN_FUTURE(threadManager, respType, callFunc) \
UNUSED(request); \
auto promise = std::make_shared<folly::Promise<respType>>(); \
respType dummyResp; \
auto f = promise->getFuture(); \
promise->setValue(std::move(dummyResp)); \
#define LOCAL_RETURN_SEMIFUTURE(threadManager, respType, callFunc) \
UNUSED(request); \
auto promise = std::make_shared<folly::Promise<respType>>(); \
respType dummyResp; \
auto f = promise->getSemiFuture(); \
promise->setValue(std::move(dummyResp)); \
return f;

namespace nebula::storage {
@@ -50,104 +50,104 @@ void GraphStorageLocalServer::stop() {
serving_ = false;
}

folly::Future<cpp2::GetNeighborsResponse> GraphStorageLocalServer::future_getNeighbors(
folly::SemiFuture<cpp2::GetNeighborsResponse> GraphStorageLocalServer::semifuture_getNeighbors(
const cpp2::GetNeighborsRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::GetNeighborsResponse, future_getNeighbors);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::GetNeighborsResponse, semifuture_getNeighbors);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_addVertices(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_addVertices(
const cpp2::AddVerticesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_addVertices);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_addVertices);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_chainAddEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_chainAddEdges(
const cpp2::AddEdgesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_chainAddEdges);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_chainAddEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_addEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_addEdges(
const cpp2::AddEdgesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_addEdges);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_addEdges);
}

folly::Future<cpp2::GetPropResponse> GraphStorageLocalServer::future_getProps(
folly::SemiFuture<cpp2::GetPropResponse> GraphStorageLocalServer::semifuture_getProps(
const cpp2::GetPropRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::GetPropResponse, future_getProps);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::GetPropResponse, semifuture_getProps);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteEdges(
const cpp2::DeleteEdgesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_deleteEdges);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_deleteEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_chainDeleteEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_chainDeleteEdges(
const cpp2::DeleteEdgesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_chainDeleteEdges);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_chainDeleteEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteVertices(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteVertices(
const cpp2::DeleteVerticesRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_deleteVertices);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_deleteVertices);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteTags(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteTags(
const cpp2::DeleteTagsRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_deleteTags);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_deleteTags);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_updateVertex(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_updateVertex(
const cpp2::UpdateVertexRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::UpdateResponse, future_updateVertex);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::UpdateResponse, semifuture_updateVertex);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_chainUpdateEdge(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_chainUpdateEdge(
const cpp2::UpdateEdgeRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::UpdateResponse, future_chainUpdateEdge);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::UpdateResponse, semifuture_chainUpdateEdge);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_updateEdge(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_updateEdge(
const cpp2::UpdateEdgeRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::UpdateResponse, future_updateEdge);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::UpdateResponse, semifuture_updateEdge);
}

folly::Future<cpp2::GetUUIDResp> GraphStorageLocalServer::future_getUUID(
folly::SemiFuture<cpp2::GetUUIDResp> GraphStorageLocalServer::semifuture_getUUID(
const cpp2::GetUUIDReq& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::GetUUIDResp, future_getUUID);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::GetUUIDResp, semifuture_getUUID);
}

folly::Future<cpp2::LookupIndexResp> GraphStorageLocalServer::future_lookupIndex(
folly::SemiFuture<cpp2::LookupIndexResp> GraphStorageLocalServer::semifuture_lookupIndex(
const cpp2::LookupIndexRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::LookupIndexResp, future_lookupIndex);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::LookupIndexResp, semifuture_lookupIndex);
}

folly::Future<cpp2::GetNeighborsResponse> GraphStorageLocalServer::future_lookupAndTraverse(
folly::SemiFuture<cpp2::GetNeighborsResponse> GraphStorageLocalServer::semifuture_lookupAndTraverse(
const cpp2::LookupAndTraverseRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::GetNeighborsResponse, future_lookupAndTraverse);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::GetNeighborsResponse, semifuture_lookupAndTraverse);
}

folly::Future<cpp2::ScanResponse> GraphStorageLocalServer::future_scanVertex(
folly::SemiFuture<cpp2::ScanResponse> GraphStorageLocalServer::semifuture_scanVertex(
const cpp2::ScanVertexRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ScanResponse, future_scanVertex);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ScanResponse, semifuture_scanVertex);
}

folly::Future<cpp2::ScanResponse> GraphStorageLocalServer::future_scanEdge(
folly::SemiFuture<cpp2::ScanResponse> GraphStorageLocalServer::semifuture_scanEdge(
const cpp2::ScanEdgeRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ScanResponse, future_scanEdge);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ScanResponse, semifuture_scanEdge);
}

folly::Future<cpp2::KVGetResponse> GraphStorageLocalServer::future_get(
folly::SemiFuture<cpp2::KVGetResponse> GraphStorageLocalServer::semifuture_get(
const cpp2::KVGetRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::KVGetResponse, future_get);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::KVGetResponse, semifuture_get);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_put(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_put(
const cpp2::KVPutRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_put);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_put);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_remove(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_remove(
const cpp2::KVRemoveRequest& request) {
LOCAL_RETURN_FUTURE(threadManager_, cpp2::ExecResponse, future_remove);
LOCAL_RETURN_SEMIFUTURE(threadManager_, cpp2::ExecResponse, semifuture_remove);
}

} // namespace nebula::storage
42 changes: 21 additions & 21 deletions src/storage/GraphStorageLocalServer.cpp
Original file line number Diff line number Diff line change
@@ -15,7 +15,7 @@ using folly::exception_wrapper;

#define LOCAL_RETURN_FUTURE(RespType, callFunc) \
auto promise = std::make_shared<folly::Promise<RespType>>(); \
auto f = promise->getFuture(); \
auto f = promise->getSemiFuture(); \
threadManager_->add([this, promise, request] { \
std::dynamic_pointer_cast<GraphStorageServiceHandler>(handler_) \
->callFunc(std::move(request)) \
@@ -58,102 +58,102 @@ void GraphStorageLocalServer::stop() {
serving_ = false;
}

folly::Future<cpp2::GetNeighborsResponse> GraphStorageLocalServer::future_getNeighbors(
folly::SemiFuture<cpp2::GetNeighborsResponse> GraphStorageLocalServer::semifuture_getNeighbors(
const cpp2::GetNeighborsRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::GetNeighborsResponse, future_getNeighbors);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_addVertices(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_addVertices(
const cpp2::AddVerticesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_addVertices);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_chainAddEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_chainAddEdges(
const cpp2::AddEdgesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_chainAddEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_addEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_addEdges(
const cpp2::AddEdgesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_addEdges);
}

folly::Future<cpp2::GetPropResponse> GraphStorageLocalServer::future_getProps(
folly::SemiFuture<cpp2::GetPropResponse> GraphStorageLocalServer::semifuture_getProps(
const cpp2::GetPropRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::GetPropResponse, future_getProps);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteEdges(
const cpp2::DeleteEdgesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_deleteEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_chainDeleteEdges(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_chainDeleteEdges(
const cpp2::DeleteEdgesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_chainDeleteEdges);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteVertices(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteVertices(
const cpp2::DeleteVerticesRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_deleteVertices);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_deleteTags(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_deleteTags(
const cpp2::DeleteTagsRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_deleteTags);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_updateVertex(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_updateVertex(
const cpp2::UpdateVertexRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::UpdateResponse, future_updateVertex);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_chainUpdateEdge(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_chainUpdateEdge(
const cpp2::UpdateEdgeRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::UpdateResponse, future_chainUpdateEdge);
}

folly::Future<cpp2::UpdateResponse> GraphStorageLocalServer::future_updateEdge(
folly::SemiFuture<cpp2::UpdateResponse> GraphStorageLocalServer::semifuture_updateEdge(
const cpp2::UpdateEdgeRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::UpdateResponse, future_updateEdge);
}

folly::Future<cpp2::GetUUIDResp> GraphStorageLocalServer::future_getUUID(
folly::SemiFuture<cpp2::GetUUIDResp> GraphStorageLocalServer::semifuture_getUUID(
const cpp2::GetUUIDReq& request) {
LOCAL_RETURN_FUTURE(cpp2::GetUUIDResp, future_getUUID);
}

folly::Future<cpp2::LookupIndexResp> GraphStorageLocalServer::future_lookupIndex(
folly::SemiFuture<cpp2::LookupIndexResp> GraphStorageLocalServer::semifuture_lookupIndex(
const cpp2::LookupIndexRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::LookupIndexResp, future_lookupIndex);
}

folly::Future<cpp2::GetNeighborsResponse> GraphStorageLocalServer::future_lookupAndTraverse(
folly::SemiFuture<cpp2::GetNeighborsResponse> GraphStorageLocalServer::semifuture_lookupAndTraverse(
const cpp2::LookupAndTraverseRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::GetNeighborsResponse, future_lookupAndTraverse);
}

folly::Future<cpp2::ScanResponse> GraphStorageLocalServer::future_scanVertex(
folly::SemiFuture<cpp2::ScanResponse> GraphStorageLocalServer::semifuture_scanVertex(
const cpp2::ScanVertexRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ScanResponse, future_scanVertex);
}

folly::Future<cpp2::ScanResponse> GraphStorageLocalServer::future_scanEdge(
folly::SemiFuture<cpp2::ScanResponse> GraphStorageLocalServer::semifuture_scanEdge(
const cpp2::ScanEdgeRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ScanResponse, future_scanEdge);
}

folly::Future<cpp2::KVGetResponse> GraphStorageLocalServer::future_get(
folly::SemiFuture<cpp2::KVGetResponse> GraphStorageLocalServer::semifuture_get(
const cpp2::KVGetRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::KVGetResponse, future_get);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_put(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_put(
const cpp2::KVPutRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_put);
}

folly::Future<cpp2::ExecResponse> GraphStorageLocalServer::future_remove(
folly::SemiFuture<cpp2::ExecResponse> GraphStorageLocalServer::semifuture_remove(
const cpp2::KVRemoveRequest& request) {
LOCAL_RETURN_FUTURE(cpp2::ExecResponse, future_remove);
}
48 changes: 28 additions & 20 deletions src/storage/GraphStorageLocalServer.h
Original file line number Diff line number Diff line change
@@ -29,31 +29,39 @@ class GraphStorageLocalServer final : public boost::noncopyable, public nebula::
void serve();

public:
folly::Future<cpp2::GetNeighborsResponse> future_getNeighbors(
folly::SemiFuture<cpp2::GetNeighborsResponse> semifuture_getNeighbors(
const cpp2::GetNeighborsRequest& request);
folly::Future<cpp2::ExecResponse> future_addVertices(const cpp2::AddVerticesRequest& request);
folly::Future<cpp2::ExecResponse> future_chainAddEdges(const cpp2::AddEdgesRequest& request);
folly::Future<cpp2::ExecResponse> future_addEdges(const cpp2::AddEdgesRequest& request);
folly::Future<cpp2::GetPropResponse> future_getProps(const cpp2::GetPropRequest& request);
folly::Future<cpp2::ExecResponse> future_deleteEdges(const cpp2::DeleteEdgesRequest& request);
folly::Future<cpp2::ExecResponse> future_chainDeleteEdges(
folly::SemiFuture<cpp2::ExecResponse> semifuture_addVertices(
const cpp2::AddVerticesRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_chainAddEdges(
const cpp2::AddEdgesRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_addEdges(const cpp2::AddEdgesRequest& request);
folly::SemiFuture<cpp2::GetPropResponse> semifuture_getProps(const cpp2::GetPropRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_deleteEdges(
const cpp2::DeleteEdgesRequest& request);
folly::Future<cpp2::ExecResponse> future_deleteVertices(
folly::SemiFuture<cpp2::ExecResponse> semifuture_chainDeleteEdges(
const cpp2::DeleteEdgesRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_deleteVertices(
const cpp2::DeleteVerticesRequest& request);
folly::Future<cpp2::ExecResponse> future_deleteTags(const cpp2::DeleteTagsRequest& request);
folly::Future<cpp2::UpdateResponse> future_updateVertex(const cpp2::UpdateVertexRequest& request);
folly::Future<cpp2::UpdateResponse> future_chainUpdateEdge(
folly::SemiFuture<cpp2::ExecResponse> semifuture_deleteTags(
const cpp2::DeleteTagsRequest& request);
folly::SemiFuture<cpp2::UpdateResponse> semifuture_updateVertex(
const cpp2::UpdateVertexRequest& request);
folly::SemiFuture<cpp2::UpdateResponse> semifuture_chainUpdateEdge(
const cpp2::UpdateEdgeRequest& request);
folly::SemiFuture<cpp2::UpdateResponse> semifuture_updateEdge(
const cpp2::UpdateEdgeRequest& request);
folly::Future<cpp2::UpdateResponse> future_updateEdge(const cpp2::UpdateEdgeRequest& request);
folly::Future<cpp2::GetUUIDResp> future_getUUID(const cpp2::GetUUIDReq& request);
folly::Future<cpp2::LookupIndexResp> future_lookupIndex(const cpp2::LookupIndexRequest& request);
folly::Future<cpp2::GetNeighborsResponse> future_lookupAndTraverse(
folly::SemiFuture<cpp2::GetUUIDResp> semifuture_getUUID(const cpp2::GetUUIDReq& request);
folly::SemiFuture<cpp2::LookupIndexResp> semifuture_lookupIndex(
const cpp2::LookupIndexRequest& request);
folly::SemiFuture<cpp2::GetNeighborsResponse> semifuture_lookupAndTraverse(
const cpp2::LookupAndTraverseRequest& request);
folly::Future<cpp2::ScanResponse> future_scanVertex(const cpp2::ScanVertexRequest& request);
folly::Future<cpp2::ScanResponse> future_scanEdge(const cpp2::ScanEdgeRequest& request);
folly::Future<cpp2::KVGetResponse> future_get(const cpp2::KVGetRequest& request);
folly::Future<cpp2::ExecResponse> future_put(const cpp2::KVPutRequest& request);
folly::Future<cpp2::ExecResponse> future_remove(const cpp2::KVRemoveRequest& request);
folly::SemiFuture<cpp2::ScanResponse> semifuture_scanVertex(
const cpp2::ScanVertexRequest& request);
folly::SemiFuture<cpp2::ScanResponse> semifuture_scanEdge(const cpp2::ScanEdgeRequest& request);
folly::SemiFuture<cpp2::KVGetResponse> semifuture_get(const cpp2::KVGetRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_put(const cpp2::KVPutRequest& request);
folly::SemiFuture<cpp2::ExecResponse> semifuture_remove(const cpp2::KVRemoveRequest& request);

private:
GraphStorageLocalServer() = default;
4 changes: 2 additions & 2 deletions tests/Makefile
Original file line number Diff line number Diff line change
@@ -25,8 +25,8 @@ PASSWORD_LOCK_TIME_IN_SECS ?= 0
# commands
gherkin_fmt = ~/.local/bin/reformat-gherkin
run_test = PYTHONPATH=$$PYTHONPATH:$(CURR_DIR)/.. $(CURR_DIR)/nebula-test-run.py
test_without_skip = python3 -m pytest -m "not skip"
test_without_skip_sa = python3 -m pytest -m "not skip and not distonly"
test_without_skip = python3 -m pytest -m "not skip" --build_dir=$(BUILD_DIR)
test_without_skip_sa = python3 -m pytest -m "not skip and not distonly" --build_dir=$(BUILD_DIR)
test_j = $(test_without_skip) -n$(J)
test_j_sa = $(test_without_skip_sa) -n$(J)

2 changes: 1 addition & 1 deletion third-party/install-gcc.sh
Original file line number Diff line number Diff line change
@@ -4,7 +4,7 @@
#
# This source code is licensed under Apache 2.0 License.

# Usage: install-gcc.sh --prefix=/opt/nebula/toolset
# Usage: install-gcc.sh --prefix=/opt

# Always use bash
shell=$(basename $(readlink /proc/$$/exe))

0 comments on commit 5814d59

Please sign in to comment.