Check memory usage before starting executor

conf/nebula-graphd.conf.default

@@ -73,3 +73,7 @@
 --enable_authorize=false
 # User login authentication type, password for nebula authentication, ldap for ldap authentication, cloud for cloud authentication
 --auth_type=password
+
+########## memory ##########
+# System memory high watermark ratio
+--system_memory_high_watermark_ratio=0.8

conf/nebula-graphd.conf.production

@@ -71,3 +71,7 @@
 --enable_authorize=false
 # User login authentication type, password for nebula authentication, ldap for ldap authentication, cloud for cloud authentication
 --auth_type=password
+
+########## memory ##########
+# System memory high watermark ratio
+--system_memory_high_watermark_ratio=0.8

resources/gflags.json

@@ -9,7 +9,8 @@
         "wal_ttl",
         "clean_wal_interval_secs",
         "custom_filter_interval_secs",
-        "accept_partial_success"
+        "accept_partial_success",
+        "system_memory_high_watermark_ratio"
     ],
     "NESTED": [
         "rocksdb_db_options",

src/executor/Executor.cpp

@@ -9,6 +9,7 @@
 #include <folly/String.h>
 #include <folly/executors/InlineExecutor.h>
 
+#include "common/base/Memory.h"
 #include "common/base/ObjectPool.h"
 #include "common/interface/gen-cpp2/graph_types.h"
 #include "context/ExecutionContext.h"
@@ -88,6 +89,7 @@
 #include "planner/plan/Mutate.h"
 #include "planner/plan/PlanNode.h"
 #include "planner/plan/Query.h"
+#include "service/GraphFlags.h"
 #include "util/ScopedTimer.h"
 
 using folly::stringPrintf;
@@ -500,6 +502,16 @@ Executor::Executor(const std::string &name, const PlanNode *node, QueryContext *
 Executor::~Executor() {}
 
 Status Executor::open() {
+    auto status = MemInfo::make();
+    NG_RETURN_IF_ERROR(status);
+    auto mem = std::move(status).value();
+    if (mem->hitsHighWatermark(FLAGS_system_memory_high_watermark_ratio)) {
+        return Status::Error(
+            "Used memory(%ldKB) hits the high watermark(%lf) of total system memory(%ldKB).",
+            mem->usedInKB(),
+            FLAGS_system_memory_high_watermark_ratio,
+            mem->totalInKB());
+    }
     numRows_ = 0;
     execTime_ = 0;
     totalDuration_.reset();

src/service/GraphFlags.cpp

@@ -52,3 +52,5 @@ DEFINE_bool(enable_optimizer, false, "Whether to enable optimizer");
 DEFINE_uint32(ft_request_retry_times, 3, "Retry times if fulltext request failed");
 
 DEFINE_bool(accept_partial_success, false, "Whether to accept partial success, default false");
+
+DEFINE_double(system_memory_high_watermark_ratio, 0.8, "high watermark ratio of system memory");

src/service/GraphFlags.h

@@ -36,6 +36,7 @@ DECLARE_bool(enable_authorize);
 DECLARE_string(auth_type);
 DECLARE_string(cloud_http_url);
 DECLARE_uint32(max_allowed_statements);
+DECLARE_double(system_memory_high_watermark_ratio);
 
 // optimizer
 DECLARE_bool(enable_optimizer);

tests/admin/test_configs.py

@@ -57,12 +57,16 @@ def test_configs(self):
 
         resp = self.client.execute('SHOW CONFIGS graph')
         self.check_resp_succeeded(resp)
-        expected_result = [['GRAPH', 'v', 'int', 'MUTABLE', v],
-                           ['GRAPH', 'minloglevel', 'int', 'MUTABLE', 0],
-                           ['GRAPH', 'slow_op_threshhold_ms', 'int', 'MUTABLE', 50],
-                           ['GRAPH', 'heartbeat_interval_secs', 'int', 'MUTABLE', 1],
-                           ['GRAPH', 'meta_client_retry_times', 'int', 'MUTABLE', 3],
-                           ['GRAPH', 'accept_partial_success', 'bool', 'MUTABLE', False]]
+
+        expected_result = [
+            ['GRAPH', 'v', 'int', 'MUTABLE', v],
+            ['GRAPH', 'minloglevel', 'int', 'MUTABLE', 0],
+            ['GRAPH', 'slow_op_threshhold_ms', 'int', 'MUTABLE', 50],
+            ['GRAPH', 'heartbeat_interval_secs', 'int', 'MUTABLE', 1],
+            ['GRAPH', 'meta_client_retry_times', 'int', 'MUTABLE', 3],
+            ['GRAPH', 'accept_partial_success', 'bool', 'MUTABLE', False],
+            ['GRAPH', 'system_memory_high_watermark_ratio', 'float', 'MUTABLE', 0.8],
+        ]
         self.check_out_of_order_result(resp, expected_result)
 
         resp = self.client.execute('SHOW CONFIGS storage')
@@ -122,5 +126,3 @@ def test_update_configs(self):
         expected_result = [['GRAPH', 'minloglevel', 'int', 'MUTABLE', 2],
                            ['STORAGE', 'minloglevel', 'int', 'MUTABLE', 3]]
         self.check_result(resp, expected_result)
-
-