diff --git a/README-CN.md b/README-CN.md
index fc2d5231815..7a2fcd7fc3b 100644
--- a/README-CN.md
+++ b/README-CN.md
@@ -8,21 +8,16 @@
-
- 
-
-
- 
-
- 
+ 
- 
+ 
+
# NebulaGraph 是什么？
@@ -108,13 +103,18 @@ NebulaGraph内核 1.x 与 2.x 数据格式、通信协议、客户端等均双
* 访问[官网](http://nebula-graph.com.cn/)
* [](https://user-images.githubusercontent.com/38887077/67449282-4362b300-f64c-11e9-878f-7efc373e5e55.jpg)
* [](https://weibo.com/p/1006067122684542/home?from=page_100606&mod=TAB#place)
-* [知乎](https://www.zhihu.com/org/nebulagraph/activities)
-* [SegmentFault](https://segmentfault.com/t/nebula)
* Email: info@vesoft.com
## 加入 NebulaGraph 社区
-[](https://github.com/vesoft-inc/nebula/discussions) [](https://discuss.nebula-graph.com.cn/) [](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) [](https://meeting.tencent.com/dm/F8NX1aRZ8PQv) [](https://calendar.google.com/calendar/u/0?cid=Z29mbGttamM3ZTVlZ2hpazI2cmNlNXVnZThAZ3JvdXAuY2FsZW5kYXIuZ29vZ2xlLmNvbQ) [](https://www.meetup.com/nebulagraph/events/287180186?utm_medium=referral&utm_campaign=share-btn_savedevents_share_modal&utm_source=link) [](https://github.com/vesoft-inc/nebula-community/wiki)
+
+
+| 加入 NebulaGraph 社区 | 加入方式 |
+| ----------------------- | ------------------------------------------------------------ |
+| 微信群 | [](https://wj.qq.com/s2/8321168/8e2f/) |
+| 提问 | [](https://discuss.nebula-graph.com.cn/) [](https://stackoverflow.com/questions/tagged/nebula-graph) [](https://github.com/vesoft-inc/nebula/discussions) |
+| 聊天 | [](https://community-chat.nebula-graph.io/) [](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) |
+| NebulaGraph Meetup 活动 | [](https://meeting.tencent.com/dm/F8NX1aRZ8PQv) [](https://calendar.google.com/calendar/u/0?cid=Z29mbGttamM3ZTVlZ2hpazI2cmNlNXVnZThAZ3JvdXAuY2FsZW5kYXIuZ29vZ2xlLmNvbQ) [](https://us02web.zoom.us/meeting/register/tZ0rcuypqDMvGdLuIm4VprTlx96wrEf062SH) [](https://www.meetup.com/nebulagraph/events/) [](https://github.com/vesoft-inc/nebula-community/wiki) |
diff --git a/README.md b/README.md
index 2ef22e51cb3..c525b32996a 100644
--- a/README.md
+++ b/README.md
@@ -4,6 +4,9 @@
A distributed, scalable, lightning-fast graph database
+
+ 
+
@@ -107,15 +110,21 @@ You can also freely deploy **NebulaGraph** as a back-end service to support your
## Contact
+* [Community Chat](https://community-chat.nebula-graph.io/)
* [Slack Channel](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g)
-* [Stack Overflow](https://stackoverflow.com/questions/tagged/nebulagraph)
+* [Stack Overflow](https://stackoverflow.com/questions/tagged/nebula-graph)
* Twitter: [@NebulaGraph](https://twitter.com/NebulaGraph)
-* [LinkedIn Page](https://www.linkedin.com/company/vesoft-nebula-graph)
+* [LinkedIn Page](https://www.linkedin.com/company/nebula-graph/)
* Email: info@vesoft.com
## Community
-[](https://github.com/vesoft-inc/nebula/discussions) [](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) [](https://us02web.zoom.us/meeting/register/tZ0rcuypqDMvGdLuIm4VprTlx96wrEf062SH) [](https://calendar.google.com/calendar/u/0?cid=Z29mbGttamM3ZTVlZ2hpazI2cmNlNXVnZThAZ3JvdXAuY2FsZW5kYXIuZ29vZ2xlLmNvbQ) [](https://www.meetup.com/nebulagraph/events/287180186?utm_medium=referral&utm_campaign=share-btn_savedevents_share_modal&utm_source=link) [](https://github.com/vesoft-inc/nebula-community/wiki) [](https://discuss.nebula-graph.com.cn/) [](https://meeting.tencent.com/dm/F8NX1aRZ8PQv)
+| Join NebulaGraph Community | Where to Find us |
+| ----------------------------------- | ------------------------------------------------------------ |
+| Asking Questions | [](https://stackoverflow.com/questions/tagged/nebula-graph) [](https://github.com/vesoft-inc/nebula/discussions) |
+| Chat with Community Members | [](https://community-chat.nebula-graph.io/) [](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) |
+| NebulaGraph Meetup | [](https://calendar.google.com/calendar/u/0?cid=Z29mbGttamM3ZTVlZ2hpazI2cmNlNXVnZThAZ3JvdXAuY2FsZW5kYXIuZ29vZ2xlLmNvbQ) [](https://us02web.zoom.us/meeting/register/tZ0rcuypqDMvGdLuIm4VprTlx96wrEf062SH) [](https://www.meetup.com/nebulagraph/events/) [](https://github.com/vesoft-inc/nebula-community/wiki) |
+| Chat, Asking, or Meeting in Chinese | [](https://wj.qq.com/s2/8321168/8e2f/) [](https://meeting.tencent.com/dm/F8NX1aRZ8PQv) [](https://discuss.nebula-graph.com.cn/) |
diff --git a/cmake/nebula/SanitizerConfig.cmake b/cmake/nebula/SanitizerConfig.cmake
index 28b5d500ef4..9f610e819a3 100644
--- a/cmake/nebula/SanitizerConfig.cmake
+++ b/cmake/nebula/SanitizerConfig.cmake
@@ -15,6 +15,7 @@ if(ENABLE_ASAN)
add_compile_options(-fsanitize=address)
add_compile_options(-g)
add_compile_options(-fno-omit-frame-pointer)
+ add_definitions(-DENABLE_ASAN)
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=address")
endif()
diff --git a/src/clients/storage/StorageClientBase-inl.h b/src/clients/storage/StorageClientBase-inl.h
index 84fef19a840..beb8d25d958 100644
--- a/src/clients/storage/StorageClientBase-inl.h
+++ b/src/clients/storage/StorageClientBase-inl.h
@@ -101,16 +101,21 @@ StorageClientBase::collectResponse(
return folly::collectAll(respFutures)
.deferValue([this, requests = std::move(requests), totalLatencies, hosts](
std::vector>>&& resps) {
+ // throw in MemoryCheckGuard verified
memory::MemoryCheckGuard guard;
StorageRpcResponse rpcResp(resps.size());
for (size_t i = 0; i < resps.size(); i++) {
auto& host = hosts->at(i);
- auto& tryResp = resps[i];
- std::optional errMsg;
+ folly::Try>& tryResp = resps[i];
if (tryResp.hasException()) {
- errMsg = std::string(tryResp.exception().what().c_str());
+ std::string errMsg = tryResp.exception().what().toStdString();
+ rpcResp.markFailure();
+ LOG(ERROR) << "There some RPC errors: " << errMsg;
+ auto req = requests.at(host);
+ auto parts = getReqPartsId(req);
+ rpcResp.appendFailedParts(parts, nebula::cpp2::ErrorCode::E_RPC_FAILURE);
} else {
- auto status = std::move(tryResp).value();
+ StatusOr status = std::move(tryResp).value();
if (status.ok()) {
auto resp = std::move(status).value();
auto result = resp.get_result();
@@ -128,17 +133,18 @@ StorageClientBase::collectResponse(
// Keep the response
rpcResp.addResponse(std::move(resp));
} else {
- errMsg = std::move(status).status().message();
+ rpcResp.markFailure();
+ Status s = std::move(status).status();
+ nebula::cpp2::ErrorCode errorCode =
+ s.code() == Status::Code::kGraphMemoryExceeded
+ ? nebula::cpp2::ErrorCode::E_GRAPH_MEMORY_EXCEEDED
+ : nebula::cpp2::ErrorCode::E_RPC_FAILURE;
+ LOG(ERROR) << "There some RPC errors: " << s.message();
+ auto req = requests.at(host);
+ auto parts = getReqPartsId(req);
+ rpcResp.appendFailedParts(parts, errorCode);
}
}
-
- if (errMsg) {
- rpcResp.markFailure();
- LOG(ERROR) << "There some RPC errors: " << errMsg.value();
- auto req = requests.at(host);
- auto parts = getReqPartsId(req);
- rpcResp.appendFailedParts(parts, nebula::cpp2::ErrorCode::E_RPC_FAILURE);
- }
}
return rpcResp;
@@ -160,12 +166,16 @@ folly::Future> StorageClientBaseclient(host, evb, false, FLAGS_storage_client_timeout_ms);
+ // Encoding invoke Cpp2Ops::write the request to protocol is in current thread,
+ // do not need to turn on in Cpp2Ops::write
return remoteFunc(client.get(), request);
})
.thenValue([spaceId, this](Response&& resp) mutable -> StatusOr {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
auto& result = resp.get_result();
for (auto& part : result.get_failed_parts()) {
@@ -196,14 +206,12 @@ folly::Future> StorageClientBase{},
- [](const std::bad_alloc&) {
- return folly::makeFuture>(std::bad_alloc());
- })
- .thenError(folly::tag_t{},
- [](const std::exception& e) {
- return folly::makeFuture>(std::runtime_error(e.what()));
- })
+ .thenError(
+ folly::tag_t{},
+ [](const std::bad_alloc&) {
+ return folly::makeFuture>(Status::GraphMemoryExceeded(
+ "(%d)", static_cast(nebula::cpp2::ErrorCode::E_GRAPH_MEMORY_EXCEEDED)));
+ })
.thenError([request, host, spaceId, this](
folly::exception_wrapper&& exWrapper) mutable -> StatusOr {
stats::StatsManager::addValue(kNumRpcSentToStoragedFailed);
diff --git a/src/codec/test/CMakeLists.txt b/src/codec/test/CMakeLists.txt
index 9b4e1272096..8c511451eaf 100644
--- a/src/codec/test/CMakeLists.txt
+++ b/src/codec/test/CMakeLists.txt
@@ -22,6 +22,8 @@ set(CODEC_TEST_LIBS
$
$
$
+ $
+ $
$
$
$
diff --git a/src/common/base/Status.cpp b/src/common/base/Status.cpp
index d80c31cf370..51d4f8698ac 100644
--- a/src/common/base/Status.cpp
+++ b/src/common/base/Status.cpp
@@ -67,10 +67,14 @@ const char *Status::toString(Code code) {
return "StatementEmpty: ";
case kSemanticError:
return "SemanticError: ";
+ case kGraphMemoryExceeded:
+ return "GraphMemoryExceeded: ";
case kKeyNotFound:
return "KeyNotFound: ";
case kPartialSuccess:
return "PartialSuccess: ";
+ case kStorageMemoryExceeded:
+ return "StorageMemoryExceeded: ";
case kSpaceNotFound:
return "SpaceNotFound: ";
case kHostNotFound:
diff --git a/src/common/base/Status.h b/src/common/base/Status.h
index 2783cb7a70b..598320ad675 100644
--- a/src/common/base/Status.h
+++ b/src/common/base/Status.h
@@ -119,12 +119,15 @@ class Status final {
// Graph engine errors
STATUS_GENERATOR(SyntaxError);
STATUS_GENERATOR(SemanticError);
+ STATUS_GENERATOR(GraphMemoryExceeded);
+
// Nothing is executed When command is comment
STATUS_GENERATOR(StatementEmpty);
// Storage engine errors
STATUS_GENERATOR(KeyNotFound);
STATUS_GENERATOR(PartialSuccess);
+ STATUS_GENERATOR(StorageMemoryExceeded);
// Meta engine errors
// TODO(dangleptr) we could use ErrorOr to replace SpaceNotFound here.
@@ -166,9 +169,11 @@ class Status final {
kSyntaxError = 201,
kStatementEmpty = 202,
kSemanticError = 203,
+ kGraphMemoryExceeded = 204,
// 3xx, for storage engine errors
kKeyNotFound = 301,
kPartialSuccess = 302,
+ kStorageMemoryExceeded = 303,
// 4xx, for meta service errors
kSpaceNotFound = 404,
kHostNotFound = 405,
diff --git a/src/common/base/test/CMakeLists.txt b/src/common/base/test/CMakeLists.txt
index 4963639cdf3..e644906d920 100644
--- a/src/common/base/test/CMakeLists.txt
+++ b/src/common/base/test/CMakeLists.txt
@@ -110,8 +110,10 @@ nebula_add_executable(
$
$
$
+ $
$
$
+ $
$
$
$
diff --git a/src/common/datatypes/DataSetOps-inl.h b/src/common/datatypes/DataSetOps-inl.h
index 2290df8bd75..89207de8497 100644
--- a/src/common/datatypes/DataSetOps-inl.h
+++ b/src/common/datatypes/DataSetOps-inl.h
@@ -13,6 +13,7 @@
#include "common/base/Base.h"
#include "common/datatypes/CommonCpp2Ops.h"
#include "common/datatypes/DataSet.h"
+#include "common/memory/MemoryTracker.h"
namespace apache {
namespace thrift {
@@ -47,7 +48,10 @@ inline constexpr protocol::TType Cpp2Ops::thriftType() {
template
uint32_t Cpp2Ops::write(Protocol* proto, nebula::DataSet const* obj) {
+ // we do not turn on memory tracker here, when the DataSet object is creating & inserting, it is
+ // in Processor::process(), where memory tracker is turned on. so we think that is enough.
uint32_t xfer = 0;
+
xfer += proto->writeStructBegin("DataSet");
xfer += proto->writeFieldBegin("column_names", protocol::T_LIST, 1);
@@ -62,11 +66,20 @@ uint32_t Cpp2Ops::write(Protocol* proto, nebula::DataSet const*
xfer += proto->writeFieldStop();
xfer += proto->writeStructEnd();
+
return xfer;
}
template
void Cpp2Ops::read(Protocol* proto, nebula::DataSet* obj) {
+ // memory usage during decode a StorageResponse should be mostly occupied
+ // by DataSet (see interface/storage.thrift), turn on memory check here.
+ //
+ // MemoryTrackerVerified:
+ // throw std::bad_alloc has verified, can be captured in
+ // StorageClientBase::getResponse's onError
+ nebula::memory::MemoryCheckGuard guard;
+
apache::thrift::detail::ProtocolReaderStructReadState readState;
readState.readStructBegin(proto);
diff --git a/src/common/datatypes/ValueOps-inl.h b/src/common/datatypes/ValueOps-inl.h
index d9ab04320f4..c58cb0a5261 100644
--- a/src/common/datatypes/ValueOps-inl.h
+++ b/src/common/datatypes/ValueOps-inl.h
@@ -103,6 +103,7 @@ template
uint32_t Cpp2Ops::write(Protocol* proto, nebula::Value const* obj) {
uint32_t xfer = 0;
xfer += proto->writeStructBegin("Value");
+ // MemoryTrackerVerified: throw bad_alloc verified
switch (obj->type()) {
case nebula::Value::Type::NULLVALUE: {
diff --git a/src/common/datatypes/test/CMakeLists.txt b/src/common/datatypes/test/CMakeLists.txt
index 0fdc5fde32c..f31731cf58b 100644
--- a/src/common/datatypes/test/CMakeLists.txt
+++ b/src/common/datatypes/test/CMakeLists.txt
@@ -43,6 +43,7 @@ nebula_add_test(
$
$
$
+ $
$
$
$
@@ -115,6 +116,7 @@ nebula_add_test(
$
$
$
+ $
$
$
LIBRARIES
diff --git a/src/common/expression/test/CMakeLists.txt b/src/common/expression/test/CMakeLists.txt
index b2884aecd67..d2c8b304859 100644
--- a/src/common/expression/test/CMakeLists.txt
+++ b/src/common/expression/test/CMakeLists.txt
@@ -128,6 +128,7 @@ nebula_add_executable(
$
$
$
+ $
$
$
$
@@ -153,6 +154,7 @@ nebula_add_executable(
$
$
$
+ $
$
$
$
diff --git a/src/common/geo/test/CMakeLists.txt b/src/common/geo/test/CMakeLists.txt
index 932fffb14e6..5b37c79ffb9 100644
--- a/src/common/geo/test/CMakeLists.txt
+++ b/src/common/geo/test/CMakeLists.txt
@@ -30,6 +30,9 @@ nebula_add_test(
$
$
$
+ $
+ $
+ $
$
$
$
diff --git a/src/common/graph/tests/CMakeLists.txt b/src/common/graph/tests/CMakeLists.txt
index cba41ef98ae..83465cf8f09 100644
--- a/src/common/graph/tests/CMakeLists.txt
+++ b/src/common/graph/tests/CMakeLists.txt
@@ -11,6 +11,9 @@ nebula_add_test(
$
$
$
+ $
+ $
+ $
$
$
$
diff --git a/src/common/id/test/CMakeLists.txt b/src/common/id/test/CMakeLists.txt
index bcd640bd625..afd71499bdc 100644
--- a/src/common/id/test/CMakeLists.txt
+++ b/src/common/id/test/CMakeLists.txt
@@ -28,6 +28,7 @@ nebula_add_executable(
$
$
$
+ $
$
$
$
@@ -67,6 +68,7 @@ nebula_add_test(
$
$
$
+ $
$
$
$
diff --git a/src/common/memory/MemoryTracker.cpp b/src/common/memory/MemoryTracker.cpp
index d7bc96432a4..548e4aacb38 100644
--- a/src/common/memory/MemoryTracker.cpp
+++ b/src/common/memory/MemoryTracker.cpp
@@ -38,6 +38,10 @@ void MemoryTracker::free(int64_t size) {
MemoryStats::instance().free(size);
}
+bool MemoryTracker::isOn() {
+ return MemoryStats::instance().throwOnMemoryExceeded();
+}
+
void MemoryTracker::allocImpl(int64_t size, bool) {
MemoryStats::instance().alloc(size);
}
diff --git a/src/common/memory/MemoryTracker.h b/src/common/memory/MemoryTracker.h
index 1d2f344d56f..f42d5223ad9 100644
--- a/src/common/memory/MemoryTracker.h
+++ b/src/common/memory/MemoryTracker.h
@@ -138,6 +138,11 @@ class MemoryStats {
threadMemoryStats_.throwOnMemoryExceeded = false;
}
+ // return true if current thread's throwOnMemoryExceeded'
+ static bool throwOnMemoryExceeded() {
+ return threadMemoryStats_.throwOnMemoryExceeded;
+ }
+
private:
inline ALWAYS_INLINE void allocGlobal(int64_t size) {
int64_t willBe = size + used_.fetch_add(size, std::memory_order_relaxed);
@@ -182,6 +187,9 @@ struct MemoryTracker {
/// This function should be called after memory deallocation.
static void free(int64_t size);
+ /// Test state of memory tracker, return true if memory tracker is turned on, otherwise false.
+ static bool isOn();
+
private:
static void allocImpl(int64_t size, bool throw_if_memory_exceeded);
};
diff --git a/src/common/memory/MemoryUtils.cpp b/src/common/memory/MemoryUtils.cpp
index aded2dd96cf..1d7170baffb 100644
--- a/src/common/memory/MemoryUtils.cpp
+++ b/src/common/memory/MemoryUtils.cpp
@@ -119,7 +119,8 @@ StatusOr MemoryUtils::hitsHighWatermark() {
}
// MemoryStats depends on jemalloc
-#if ENABLE_JEMALLOC
+#ifdef ENABLE_JEMALLOC
+#ifndef ENABLE_ASAN
// set MemoryStats limit (MemoryTracker track-able memory)
int64_t trackable = total - FLAGS_memory_tracker_untracked_reserved_memory_mb * MiB;
if (trackable > 0) {
@@ -134,16 +135,9 @@ StatusOr MemoryUtils::hitsHighWatermark() {
if (FLAGS_memory_purge_enabled) {
int64_t now = time::WallClock::fastNowInSec();
if (now - kLastPurge_ > FLAGS_memory_purge_interval_seconds) {
- // mallctl seems has issue with address_sanitizer, do purge only when address_sanitizer is off
-#if defined(__clang__)
-#if defined(__has_feature)
-#if not __has_feature(address_sanitizer)
- mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0);
-#endif
-#endif
-#else // gcc
+ // jemalloc seems has issue with address_sanitizer, do purge only when address_sanitizer is
+ // off
mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0);
-#endif
kLastPurge_ = now;
}
}
@@ -173,7 +167,7 @@ StatusOr MemoryUtils::hitsHighWatermark() {
kLastPrintMemoryTrackerStats_ = now;
}
}
-
+#endif
#endif
auto hits = (1 - available / total) > FLAGS_system_memory_high_watermark_ratio;
diff --git a/src/common/memory/NewDelete.cpp b/src/common/memory/NewDelete.cpp
index 4ff129cf331..5bc67076553 100644
--- a/src/common/memory/NewDelete.cpp
+++ b/src/common/memory/NewDelete.cpp
@@ -11,23 +11,16 @@
/// Two condition need check before MemoryTracker is on
/// 1. jemalloc is used
/// MemoryTracker need jemalloc API to get accurate size of alloc/free memory.
-#if ENABLE_JEMALLOC
/// 2. address_sanitizer is off
/// sanitizer has already override the new/delete operator,
/// only override new/delete operator only when address_sanitizer is off
-#if defined(__clang__)
-#if defined(__has_feature)
-#if not __has_feature(address_sanitizer)
+#ifdef ENABLE_JEMALLOC
+#ifndef ENABLE_ASAN
#define ENABLE_MEMORY_TRACKER
#endif
#endif
-#else // gcc
-#define ENABLE_MEMORY_TRACKER
-#endif
-#endif
-
-#if defined(ENABLE_MEMORY_TRACKER)
+#ifdef ENABLE_MEMORY_TRACKER
/// new
void *operator new(std::size_t size) {
nebula::memory::trackMemory(size);
diff --git a/src/common/utils/test/CMakeLists.txt b/src/common/utils/test/CMakeLists.txt
index afbdc3e9de4..63727036237 100644
--- a/src/common/utils/test/CMakeLists.txt
+++ b/src/common/utils/test/CMakeLists.txt
@@ -21,6 +21,9 @@ nebula_add_test(
$
$
$
+ $
+ $
+ $
$
$
LIBRARIES
@@ -47,6 +50,9 @@ nebula_add_test(
$
$
$
+ $
+ $
+ $
$
$
LIBRARIES
@@ -76,6 +82,9 @@ nebula_add_test(
$
$
$
+ $
+ $
+ $
$
$
LIBRARIES
@@ -94,6 +103,8 @@ nebula_add_test(
$
$
$
+ $
+ $
$
$
$
diff --git a/src/graph/executor/Executor.cpp b/src/graph/executor/Executor.cpp
index e3d62265b8b..313d8cd7f59 100644
--- a/src/graph/executor/Executor.cpp
+++ b/src/graph/executor/Executor.cpp
@@ -603,6 +603,8 @@ Status Executor::open() {
}
Status Executor::close() {
+ // MemoryTrackerVerified
+
ProfilingStats stats;
stats.totalDurationInUs = totalDuration_.elapsedInUSec();
stats.rows = numRows_;
@@ -725,6 +727,7 @@ bool Executor::movable(const Variable *var) {
}
Status Executor::finish(Result &&result) {
+ // MemoryTrackerVerified
if (!FLAGS_enable_lifetime_optimize ||
node()->outputVarPtr()->userCount.load(std::memory_order_relaxed) != 0) {
numRows_ = result.size();
diff --git a/src/graph/executor/Executor.h b/src/graph/executor/Executor.h
index 670eb4df302..ba9f90970cd 100644
--- a/src/graph/executor/Executor.h
+++ b/src/graph/executor/Executor.h
@@ -77,8 +77,8 @@ class Executor : private boost::noncopyable, private cpp::NonMovable {
folly::Future error(Status status) const;
static Status memoryExceededStatus() {
- return Status::Error("Graph Error: GRAPH_MEMORY_EXCEEDED(%d)",
- static_cast(nebula::cpp2::ErrorCode::E_GRAPH_MEMORY_EXCEEDED));
+ return Status::GraphMemoryExceeded(
+ "(%d)", static_cast(nebula::cpp2::ErrorCode::E_GRAPH_MEMORY_EXCEEDED));
}
protected:
@@ -159,6 +159,7 @@ auto Executor::runMultiJobs(ScatterFunc &&scatter, GatherFunc &&gather, Iterator
futures.emplace_back(folly::via(
runner(),
[begin, end, tmpIter = iter->copy(), f = std::move(scatter)]() mutable -> ScatterResult {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
// Since not all iterators are linear, so iterates to the begin pos
size_t tmp = 0;
@@ -173,17 +174,7 @@ auto Executor::runMultiJobs(ScatterFunc &&scatter, GatherFunc &&gather, Iterator
}
// Gather all results and do post works
- return folly::collect(futures)
- .via(runner())
- .thenValue(std::move(gather))
- .thenError(folly::tag_t{},
- [](const std::bad_alloc &) {
- return folly::makeFuture(std::runtime_error(
- "Memory Limit Exceeded, " + memory::MemoryStats::instance().toString()));
- })
- .thenError(folly::tag_t{}, [](const std::exception &e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ return folly::collect(futures).via(runner()).thenValue(std::move(gather));
}
} // namespace graph
} // namespace nebula
diff --git a/src/graph/executor/StorageAccessExecutor.h b/src/graph/executor/StorageAccessExecutor.h
index 18ddbe83e40..a3881325f7c 100644
--- a/src/graph/executor/StorageAccessExecutor.h
+++ b/src/graph/executor/StorageAccessExecutor.h
@@ -131,9 +131,11 @@ class StorageAccessExecutor : public Executor {
"Storage Error: Part {} raft buffer is full. Please retry later.", partId));
case nebula::cpp2::ErrorCode::E_RAFT_ATOMIC_OP_FAILED:
return Status::Error("Storage Error: Atomic operation failed.");
+ // E_GRAPH_MEMORY_EXCEEDED may happen during rpc response deserialize.
+ case nebula::cpp2::ErrorCode::E_GRAPH_MEMORY_EXCEEDED:
+ return Status::GraphMemoryExceeded("(%d)", static_cast(code));
case nebula::cpp2::ErrorCode::E_STORAGE_MEMORY_EXCEEDED:
- return Status::Error("Storage Error: STORAGE_MEMORY_EXCEEDED(%d)",
- static_cast(code));
+ return Status::StorageMemoryExceeded("(%d)", static_cast(code));
default:
auto status = Status::Error("Storage Error: part: %d, error: %s(%d).",
partId,
diff --git a/src/graph/executor/algo/BFSShortestPathExecutor.cpp b/src/graph/executor/algo/BFSShortestPathExecutor.cpp
index 7cb4fda3447..cf9faafa0d9 100644
--- a/src/graph/executor/algo/BFSShortestPathExecutor.cpp
+++ b/src/graph/executor/algo/BFSShortestPathExecutor.cpp
@@ -10,6 +10,8 @@ DECLARE_int32(num_operator_threads);
namespace nebula {
namespace graph {
folly::Future BFSShortestPathExecutor::execute() {
+ // MemoryTrackerVerified
+
SCOPED_TIMER(&execTime_);
pathNode_ = asNode(node());
terminateEarlyVar_ = pathNode_->terminateEarlyVar();
@@ -30,10 +32,12 @@ folly::Future BFSShortestPathExecutor::execute() {
std::vector> futures;
auto leftFuture = folly::via(runner(), [this]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return buildPath(false);
});
auto rightFuture = folly::via(runner(), [this]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return buildPath(true);
});
@@ -111,6 +115,7 @@ Status BFSShortestPathExecutor::buildPath(bool reverse) {
currentEdges.emplace(std::move(dst), std::move(edge));
}
}
+
// set nextVid
const auto& nextVidVar = reverse ? pathNode_->rightVidVar() : pathNode_->leftVidVar();
ectx_->setResult(nextVidVar, ResultBuilder().value(std::move(nextStepVids)).build());
@@ -170,12 +175,6 @@ folly::Future BFSShortestPathExecutor::conjunctPath() {
currentDs_.append(std::move(resp));
}
return Status::OK();
- })
- .thenError(
- folly::tag_t{},
- [](const std::bad_alloc&) { return folly::makeFuture(memoryExceededStatus()); })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
diff --git a/src/graph/executor/algo/BatchShortestPath.cpp b/src/graph/executor/algo/BatchShortestPath.cpp
index 2400364b592..95ac030019d 100644
--- a/src/graph/executor/algo/BatchShortestPath.cpp
+++ b/src/graph/executor/algo/BatchShortestPath.cpp
@@ -15,6 +15,7 @@ namespace graph {
folly::Future BatchShortestPath::execute(const HashSet& startVids,
const HashSet& endVids,
DataSet* result) {
+ // MemoryTrackerVerified
size_t rowSize = init(startVids, endVids);
std::vector> futures;
futures.reserve(rowSize);
@@ -22,26 +23,18 @@ folly::Future BatchShortestPath::execute(const HashSet& startVids,
resultDs_[rowNum].colNames = pathNode_->colNames();
futures.emplace_back(shortestPath(rowNum, 1));
}
- return folly::collect(futures)
- .via(qctx_->rctx()->runner())
- .thenValue([this, result](auto&& resps) {
- memory::MemoryCheckGuard guard;
- for (auto& resp : resps) {
- NG_RETURN_IF_ERROR(resp);
- }
- result->colNames = pathNode_->colNames();
- for (auto& ds : resultDs_) {
- result->append(std::move(ds));
- }
- return Status::OK();
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ return folly::collect(futures).via(runner()).thenValue([this, result](auto&& resps) {
+ // MemoryTrackerVerified
+ memory::MemoryCheckGuard guard;
+ for (auto& resp : resps) {
+ NG_RETURN_IF_ERROR(resp);
+ }
+ result->colNames = pathNode_->colNames();
+ for (auto& ds : resultDs_) {
+ result->append(std::move(ds));
+ }
+ return Status::OK();
+ });
}
size_t BatchShortestPath::init(const HashSet& startVids, const HashSet& endVids) {
@@ -106,8 +99,9 @@ folly::Future BatchShortestPath::shortestPath(size_t rowNum, size_t step
futures.emplace_back(getNeighbors(rowNum, stepNum, false));
futures.emplace_back(getNeighbors(rowNum, stepNum, true));
return folly::collect(futures)
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum, stepNum](auto&& resps) {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
for (auto& resp : resps) {
if (!resp.ok()) {
@@ -116,6 +110,10 @@ folly::Future BatchShortestPath::shortestPath(size_t rowNum, size_t step
}
return handleResponse(rowNum, stepNum);
})
+ // This thenError is necessary to catch bad_alloc, seems the returned future
+ // is related to two routines: getNeighbors, handleResponse, each of them launch some task in
+ // separate thread, if any one of routine throw bad_alloc, fail the query, will cause another
+ // to run on a maybe already released BatchShortestPath object
.thenError(folly::tag_t{},
[](const std::bad_alloc&) {
return folly::makeFuture(Executor::memoryExceededStatus());
@@ -151,18 +149,12 @@ folly::Future BatchShortestPath::getNeighbors(size_t rowNum, size_t step
-1,
nullptr,
nullptr)
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum, reverse, stepNum, getNbrTime](auto&& resp) {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
addStats(resp, stepNum, getNbrTime.elapsedInUSec(), reverse);
return buildPath(rowNum, std::move(resp), reverse);
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
@@ -281,14 +273,19 @@ Status BatchShortestPath::doBuildPath(size_t rowNum, GetNeighborsIter* iter, boo
folly::Future BatchShortestPath::handleResponse(size_t rowNum, size_t stepNum) {
return folly::makeFuture(Status::OK())
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum](auto&& status) {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
+
// odd step
UNUSED(status);
return conjunctPath(rowNum, true);
})
.thenValue([this, rowNum, stepNum](auto&& terminate) {
+ // MemoryTrackerVerified
+ memory::MemoryCheckGuard guard;
+
// even Step
if (terminate || stepNum * 2 > maxStep_) {
return folly::makeFuture(true);
@@ -296,6 +293,9 @@ folly::Future BatchShortestPath::handleResponse(size_t rowNum, size_t st
return conjunctPath(rowNum, false);
})
.thenValue([this, rowNum, stepNum](auto&& result) {
+ // MemoryTrackerVerified
+ memory::MemoryCheckGuard guard;
+
if (result || stepNum * 2 >= maxStep_) {
return folly::makeFuture(Status::OK());
}
@@ -319,13 +319,6 @@ folly::Future BatchShortestPath::handleResponse(size_t rowNum, size_t st
leftPathMap.clear();
rightPathMap.clear();
return shortestPath(rowNum, stepNum + 1);
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
@@ -379,64 +372,61 @@ folly::Future BatchShortestPath::conjunctPath(size_t rowNum, bool oddStep)
}
auto future = getMeetVids(rowNum, oddStep, meetVids);
- return future.via(qctx_->rctx()->runner())
- .thenValue([this, rowNum, oddStep](auto&& vertices) {
- memory::MemoryCheckGuard guard;
- if (vertices.empty()) {
- return false;
- }
- robin_hood::unordered_flat_map> verticesMap;
- for (auto& vertex : vertices) {
- verticesMap[vertex.getVertex().vid] = std::move(vertex);
+ return future.via(runner()).thenValue([this, rowNum, oddStep](auto&& vertices) {
+ // MemoryTrackerVerified
+ memory::MemoryCheckGuard guard;
+
+ if (vertices.empty()) {
+ return false;
+ }
+ robin_hood::unordered_flat_map> verticesMap;
+ for (auto& vertex : vertices) {
+ verticesMap[vertex.getVertex().vid] = std::move(vertex);
+ }
+ auto& terminationMap = terminationMaps_[rowNum];
+ auto& leftPathMaps = currentLeftPathMaps_[rowNum];
+ auto& rightPathMaps = oddStep ? preRightPathMaps_[rowNum] : currentRightPathMaps_[rowNum];
+ for (const auto& leftPathMap : leftPathMaps) {
+ auto findCommonVid = rightPathMaps.find(leftPathMap.first);
+ if (findCommonVid == rightPathMaps.end()) {
+ continue;
+ }
+ auto findCommonVertex = verticesMap.find(findCommonVid->first);
+ if (findCommonVertex == verticesMap.end()) {
+ continue;
+ }
+ auto& rightPaths = findCommonVid->second;
+ for (const auto& srcPaths : leftPathMap.second) {
+ auto range = terminationMap.equal_range(srcPaths.first);
+ if (range.first == range.second) {
+ continue;
}
- auto& terminationMap = terminationMaps_[rowNum];
- auto& leftPathMaps = currentLeftPathMaps_[rowNum];
- auto& rightPathMaps = oddStep ? preRightPathMaps_[rowNum] : currentRightPathMaps_[rowNum];
- for (const auto& leftPathMap : leftPathMaps) {
- auto findCommonVid = rightPathMaps.find(leftPathMap.first);
- if (findCommonVid == rightPathMaps.end()) {
- continue;
- }
- auto findCommonVertex = verticesMap.find(findCommonVid->first);
- if (findCommonVertex == verticesMap.end()) {
- continue;
- }
- auto& rightPaths = findCommonVid->second;
- for (const auto& srcPaths : leftPathMap.second) {
- auto range = terminationMap.equal_range(srcPaths.first);
- if (range.first == range.second) {
- continue;
- }
- for (const auto& dstPaths : rightPaths) {
- for (auto found = range.first; found != range.second; ++found) {
- if (found->second.first == dstPaths.first) {
- if (singleShortest_ && !found->second.second) {
- break;
- }
- doConjunctPath(
- srcPaths.second, dstPaths.second, findCommonVertex->second, rowNum);
- found->second.second = false;
- }
+ for (const auto& dstPaths : rightPaths) {
+ for (auto found = range.first; found != range.second; ++found) {
+ if (found->second.first == dstPaths.first) {
+ if (singleShortest_ && !found->second.second) {
+ break;
}
+ doConjunctPath(srcPaths.second, dstPaths.second, findCommonVertex->second, rowNum);
+ found->second.second = false;
}
}
}
- // update terminationMap
- for (auto iter = terminationMap.begin(); iter != terminationMap.end();) {
- if (!iter->second.second) {
- iter = terminationMap.erase(iter);
- } else {
- ++iter;
- }
- }
- if (terminationMap.empty()) {
- return true;
- }
- return false;
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ }
+ }
+ // update terminationMap
+ for (auto iter = terminationMap.begin(); iter != terminationMap.end();) {
+ if (!iter->second.second) {
+ iter = terminationMap.erase(iter);
+ } else {
+ ++iter;
+ }
+ }
+ if (terminationMap.empty()) {
+ return true;
+ }
+ return false;
+ });
}
void BatchShortestPath::doConjunctPath(const std::vector& leftPaths,
diff --git a/src/graph/executor/algo/MultiShortestPathExecutor.cpp b/src/graph/executor/algo/MultiShortestPathExecutor.cpp
index f77dc4d9d59..d9f0fddff15 100644
--- a/src/graph/executor/algo/MultiShortestPathExecutor.cpp
+++ b/src/graph/executor/algo/MultiShortestPathExecutor.cpp
@@ -46,6 +46,7 @@ folly::Future MultiShortestPathExecutor::execute() {
return conjunctPath(false);
})
.thenValue([this](auto&& resp) {
+ memory::MemoryCheckGuard guard;
UNUSED(resp);
preRightPaths_ = rightPaths_;
// update history
@@ -311,30 +312,25 @@ folly::Future MultiShortestPathExecutor::conjunctPath(bool oddStep) {
futures.emplace_back(std::move(future));
}
- return folly::collect(futures)
- .via(runner())
- .thenValue([this](auto&& resps) {
- memory::MemoryCheckGuard guard;
- for (auto& resp : resps) {
- currentDs_.append(std::move(resp));
- }
+ return folly::collect(futures).via(runner()).thenValue([this](auto&& resps) {
+ memory::MemoryCheckGuard guard;
+ for (auto& resp : resps) {
+ currentDs_.append(std::move(resp));
+ }
- for (auto iter = terminationMap_.begin(); iter != terminationMap_.end();) {
- if (!iter->second.second) {
- iter = terminationMap_.erase(iter);
- } else {
- ++iter;
- }
- }
- if (terminationMap_.empty()) {
- ectx_->setValue(terminationVar_, true);
- return true;
- }
- return false;
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ for (auto iter = terminationMap_.begin(); iter != terminationMap_.end();) {
+ if (!iter->second.second) {
+ iter = terminationMap_.erase(iter);
+ } else {
+ ++iter;
+ }
+ }
+ if (terminationMap_.empty()) {
+ ectx_->setValue(terminationVar_, true);
+ return true;
+ }
+ return false;
+ });
}
void MultiShortestPathExecutor::setNextStepVid(const Interims& paths, const string& var) {
diff --git a/src/graph/executor/algo/ProduceAllPathsExecutor.cpp b/src/graph/executor/algo/ProduceAllPathsExecutor.cpp
index 44e771ab7cd..f2b709a000f 100644
--- a/src/graph/executor/algo/ProduceAllPathsExecutor.cpp
+++ b/src/graph/executor/algo/ProduceAllPathsExecutor.cpp
@@ -28,10 +28,12 @@ folly::Future ProduceAllPathsExecutor::execute() {
}
std::vector> futures;
auto leftFuture = folly::via(runner(), [this]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return buildPath(false);
});
auto rightFuture = folly::via(runner(), [this]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return buildPath(true);
});
@@ -156,12 +158,14 @@ folly::Future ProduceAllPathsExecutor::conjunctPath() {
auto endIter = leftIter;
endIter++;
auto oddStepFuture = folly::via(runner(), [this, startIter, endIter]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return doConjunct(startIter, endIter, true);
});
futures.emplace_back(std::move(oddStepFuture));
if (step_ * 2 <= pathNode_->steps()) {
auto evenStepFuture = folly::via(runner(), [this, startIter, endIter]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return doConjunct(startIter, endIter, false);
});
@@ -175,12 +179,14 @@ folly::Future ProduceAllPathsExecutor::conjunctPath() {
if (i != 0) {
auto endIter = leftPaths_.end();
auto oddStepFuture = folly::via(runner(), [this, startIter, endIter]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return doConjunct(startIter, endIter, true);
});
futures.emplace_back(std::move(oddStepFuture));
if (step_ * 2 <= pathNode_->steps()) {
auto evenStepFuture = folly::via(runner(), [this, startIter, endIter]() {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
return doConjunct(startIter, endIter, false);
});
@@ -200,12 +206,6 @@ folly::Future ProduceAllPathsExecutor::conjunctPath() {
leftPaths_.clear();
rightPaths_.clear();
return Status::OK();
- })
- .thenError(
- folly::tag_t{},
- [](const std::bad_alloc&) { return folly::makeFuture(memoryExceededStatus()); })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
diff --git a/src/graph/executor/algo/ShortestPathBase.cpp b/src/graph/executor/algo/ShortestPathBase.cpp
index 9bc82ed0ed9..e809f27f72b 100644
--- a/src/graph/executor/algo/ShortestPathBase.cpp
+++ b/src/graph/executor/algo/ShortestPathBase.cpp
@@ -197,5 +197,12 @@ void ShortestPathBase::addStats(PropRpcResponse& resp, int64_t timeInUSec) const
statsLock_.unlock();
}
+folly::Executor* ShortestPathBase::runner() const {
+ if (!qctx_ || !qctx_->rctx() || !qctx_->rctx()->runner()) {
+ return &folly::InlineExecutor::instance();
+ }
+ return qctx_->rctx()->runner();
+}
+
} // namespace graph
} // namespace nebula
diff --git a/src/graph/executor/algo/ShortestPathBase.h b/src/graph/executor/algo/ShortestPathBase.h
index b52767a9d6e..d107f7fbbcd 100644
--- a/src/graph/executor/algo/ShortestPathBase.h
+++ b/src/graph/executor/algo/ShortestPathBase.h
@@ -51,6 +51,8 @@ class ShortestPathBase {
void addStats(PropRpcResponse& resp, int64_t timeInUSec) const;
+ folly::Executor* runner() const;
+
template
StatusOr handleCompleteness(const storage::StorageRpcResponse& rpcResp,
bool isPartialSuccessAccepted) const {
diff --git a/src/graph/executor/algo/ShortestPathExecutor.cpp b/src/graph/executor/algo/ShortestPathExecutor.cpp
index d8269ae074f..98381a9e72a 100644
--- a/src/graph/executor/algo/ShortestPathExecutor.cpp
+++ b/src/graph/executor/algo/ShortestPathExecutor.cpp
@@ -3,6 +3,7 @@
// This source code is licensed under Apache 2.0 License.
#include "graph/executor/algo/ShortestPathExecutor.h"
+#include "common/memory/MemoryTracker.h"
#include "graph/executor/algo/BatchShortestPath.h"
#include "graph/executor/algo/SingleShortestPath.h"
#include "graph/service/GraphFlags.h"
@@ -15,6 +16,7 @@ DEFINE_uint32(num_path_thread, 0, "number of concurrent threads when do shortest
namespace nebula {
namespace graph {
folly::Future ShortestPathExecutor::execute() {
+ // MemoryTrackerVerified
SCOPED_TIMER(&execTime_);
if (FLAGS_num_path_thread == 0) {
FLAGS_num_path_thread = get_nprocs_conf();
diff --git a/src/graph/executor/algo/SingleShortestPath.cpp b/src/graph/executor/algo/SingleShortestPath.cpp
index 69d3dba483a..fd509d765f6 100644
--- a/src/graph/executor/algo/SingleShortestPath.cpp
+++ b/src/graph/executor/algo/SingleShortestPath.cpp
@@ -21,26 +21,17 @@ folly::Future SingleShortestPath::execute(const HashSet& startVids,
resultDs_[rowNum].colNames = pathNode_->colNames();
futures.emplace_back(shortestPath(rowNum, 1));
}
- return folly::collect(futures)
- .via(qctx_->rctx()->runner())
- .thenValue([this, result](auto&& resps) {
- memory::MemoryCheckGuard guard;
- for (auto& resp : resps) {
- NG_RETURN_IF_ERROR(resp);
- }
- result->colNames = pathNode_->colNames();
- for (auto& ds : resultDs_) {
- result->append(std::move(ds));
- }
- return Status::OK();
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ return folly::collect(futures).via(runner()).thenValue([this, result](auto&& resps) {
+ memory::MemoryCheckGuard guard;
+ for (auto& resp : resps) {
+ NG_RETURN_IF_ERROR(resp);
+ }
+ result->colNames = pathNode_->colNames();
+ for (auto& ds : resultDs_) {
+ result->append(std::move(ds));
+ }
+ return Status::OK();
+ });
}
void SingleShortestPath::init(const HashSet& startVids, const HashSet& endVids, size_t rowSize) {
@@ -75,7 +66,7 @@ folly::Future SingleShortestPath::shortestPath(size_t rowNum, size_t ste
futures.emplace_back(getNeighbors(rowNum, stepNum, false));
futures.emplace_back(getNeighbors(rowNum, stepNum, true));
return folly::collect(futures)
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum, stepNum](auto&& resps) {
memory::MemoryCheckGuard guard;
for (auto& resp : resps) {
@@ -122,18 +113,11 @@ folly::Future SingleShortestPath::getNeighbors(size_t rowNum,
-1,
nullptr,
nullptr)
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum, stepNum, getNbrTime, reverse](auto&& resp) {
memory::MemoryCheckGuard guard;
addStats(resp, stepNum, getNbrTime.elapsedInUSec(), reverse);
return buildPath(rowNum, std::move(resp), reverse);
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
@@ -198,13 +182,16 @@ Status SingleShortestPath::doBuildPath(size_t rowNum, GetNeighborsIter* iter, bo
folly::Future SingleShortestPath::handleResponse(size_t rowNum, size_t stepNum) {
return folly::makeFuture(Status::OK())
- .via(qctx_->rctx()->runner())
+ .via(runner())
.thenValue([this, rowNum, stepNum](auto&& status) {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
+
UNUSED(status);
return conjunctPath(rowNum, stepNum);
})
.thenValue([this, rowNum, stepNum](auto&& result) {
+ memory::MemoryCheckGuard guard;
if (result || stepNum * 2 >= maxStep_) {
return folly::makeFuture(Status::OK());
}
@@ -214,13 +201,6 @@ folly::Future SingleShortestPath::handleResponse(size_t rowNum, size_t s
return folly::makeFuture(Status::OK());
}
return shortestPath(rowNum, stepNum + 1);
- })
- .thenError(folly::tag_t{},
- [](const std::bad_alloc&) {
- return folly::makeFuture(Executor::memoryExceededStatus());
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
@@ -281,38 +261,36 @@ void SingleShortestPath::buildOddPath(size_t rowNum, const std::vector& m
folly::Future SingleShortestPath::buildEvenPath(size_t rowNum,
const std::vector& meetVids) {
auto future = getMeetVidsProps(meetVids);
- return future.via(qctx_->rctx()->runner())
- .thenValue([this, rowNum](auto&& vertices) {
- memory::MemoryCheckGuard guard;
- if (vertices.empty()) {
- return false;
- }
- for (auto& meetVertex : vertices) {
- if (!meetVertex.isVertex()) {
- continue;
- }
- auto meetVid = meetVertex.getVertex().vid;
- auto leftPaths = createLeftPath(rowNum, meetVid);
- auto rightPaths = createRightPath(rowNum, meetVid, false);
- for (auto& leftPath : leftPaths) {
- for (auto& rightPath : rightPaths) {
- Row path = leftPath;
- auto& steps = path.values.back().mutableList().values;
- steps.emplace_back(meetVertex);
- steps.insert(steps.end(), rightPath.values.begin(), rightPath.values.end() - 1);
- path.emplace_back(rightPath.values.back());
- resultDs_[rowNum].rows.emplace_back(std::move(path));
- if (singleShortest_) {
- return true;
- }
- }
+ return future.via(runner()).thenValue([this, rowNum](auto&& vertices) {
+ // MemoryTrackerVerified
+ memory::MemoryCheckGuard guard;
+
+ if (vertices.empty()) {
+ return false;
+ }
+ for (auto& meetVertex : vertices) {
+ if (!meetVertex.isVertex()) {
+ continue;
+ }
+ auto meetVid = meetVertex.getVertex().vid;
+ auto leftPaths = createLeftPath(rowNum, meetVid);
+ auto rightPaths = createRightPath(rowNum, meetVid, false);
+ for (auto& leftPath : leftPaths) {
+ for (auto& rightPath : rightPaths) {
+ Row path = leftPath;
+ auto& steps = path.values.back().mutableList().values;
+ steps.emplace_back(meetVertex);
+ steps.insert(steps.end(), rightPath.values.begin(), rightPath.values.end() - 1);
+ path.emplace_back(rightPath.values.back());
+ resultDs_[rowNum].rows.emplace_back(std::move(path));
+ if (singleShortest_) {
+ return true;
}
}
- return true;
- })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
- });
+ }
+ }
+ return true;
+ });
}
std::vector SingleShortestPath::createLeftPath(size_t rowNum, const Value& meetVid) {
diff --git a/src/graph/executor/algo/SubgraphExecutor.cpp b/src/graph/executor/algo/SubgraphExecutor.cpp
index e51f4853c0c..cb3e9ca7634 100644
--- a/src/graph/executor/algo/SubgraphExecutor.cpp
+++ b/src/graph/executor/algo/SubgraphExecutor.cpp
@@ -53,6 +53,7 @@ folly::Future SubgraphExecutor::getNeighbors() {
currentStep_ == 1 ? nullptr : subgraph_->tagFilter())
.via(runner())
.thenValue([this, getNbrTime](RpcResponse&& resp) mutable {
+ // MemoryTrackerVerified
memory::MemoryCheckGuard guard;
otherStats_.emplace("total_rpc_time", folly::sformat("{}(us)", getNbrTime.elapsedInUSec()));
auto& hostLatency = resp.hostLatency();
@@ -67,12 +68,6 @@ folly::Future SubgraphExecutor::getNeighbors() {
}
vids_.clear();
return handleResponse(std::move(resp));
- })
- .thenError(
- folly::tag_t{},
- [](const std::bad_alloc&) { return folly::makeFuture(memoryExceededStatus()); })
- .thenError(folly::tag_t{}, [](const std::exception& e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
diff --git a/src/graph/executor/logic/ArgumentExecutor.cpp b/src/graph/executor/logic/ArgumentExecutor.cpp
index eab46383f17..1e74f3ba165 100644
--- a/src/graph/executor/logic/ArgumentExecutor.cpp
+++ b/src/graph/executor/logic/ArgumentExecutor.cpp
@@ -13,6 +13,7 @@ ArgumentExecutor::ArgumentExecutor(const PlanNode *node, QueryContext *qctx)
: Executor("ArgumentExecutor", node, qctx) {}
folly::Future ArgumentExecutor::execute() {
+ // MemoryTrackerVerified
auto *argNode = asNode(node());
auto &alias = argNode->getAlias();
auto iter = ectx_->getResult(argNode->inputVar()).iter();
diff --git a/src/graph/executor/maintain/EdgeExecutor.cpp b/src/graph/executor/maintain/EdgeExecutor.cpp
index e198624af6a..1c8a3151f44 100644
--- a/src/graph/executor/maintain/EdgeExecutor.cpp
+++ b/src/graph/executor/maintain/EdgeExecutor.cpp
@@ -21,18 +21,13 @@ folly::Future CreateEdgeExecutor::execute() {
.via(runner())
.thenValue([ceNode, spaceId](StatusOr resp) {
memory::MemoryCheckGuard guard;
+ // throw in MemoryCheckGuard verified
if (!resp.ok()) {
LOG(WARNING) << "SpaceId: " << spaceId << ", Create edge `" << ceNode->getName()
<< "' failed: " << resp.status();
return resp.status();
}
return Status::OK();
- })
- .thenError(
- folly::tag_t{},
- [](const std::bad_alloc &) { return folly::makeFuture(memoryExceededStatus()); })
- .thenError(folly::tag_t{}, [](const std::exception &e) {
- return folly::makeFuture(std::runtime_error(e.what()));
});
}
@@ -47,6 +42,7 @@ folly::Future DescEdgeExecutor::execute() {
.via(runner())
.thenValue([this, deNode, spaceId](StatusOr resp) {
memory::MemoryCheckGuard guard;
+ // MemoryTrackerVerified
if (!resp.ok()) {
LOG(WARNING) << "SpaceId: " << spaceId << ", Desc edge `" << deNode->getName()
<< "' failed: " << resp.status();
@@ -62,12 +58,6 @@ folly::Future DescEdgeExecutor::execute() {
.value(Value(std::move(ret).value()))
.iter(Iterator::Kind::kDefault)
.build());
- })
- .thenError(
- folly::tag_t{},
- [](const std::bad_alloc &) { return folly::makeFuture
