support user define memory usage

Signed-off-by: Lei Wang <[email protected]>

charts/gart/templates/writer/statefulset.yaml

@@ -53,6 +53,15 @@ spec:
           --enable-bulkload ${ENABLE_BULKLOAD} \
           --rg-from-etcd true \
           --k8s-mode true \
+          {{- if .Values.dataconfig.defaultMaxVertexNum }}
+          --default-max-vertex-num {{ .Values.dataconfig.defaultMaxVertexNum }} \
+          {{- end }}
+          {{- if .Values.dataconfig.defaultMaxMemUsage }}
+          --default-max-mem-usage {{ .Values.dataconfig.defaultMaxMemUsage }} \
+          {{- end }}
+          {{- if .Values.dataconfig.customMemConfig }}
+          --customized-memory-config {{ .Values.dataconfig.customMemConfig }} \
+          {{- end }}
           --role writer &&
           sleep infinity
         envFrom:

charts/gart/values.yaml

@@ -250,3 +250,6 @@ dataconfig:
   # secondsPerEpoch: 60
   # if you run GAE task, set useGAE to 1, otherwise run GIE task. Default run GIE task
   useGAE: 0
+  # defaultMaxVertexNum: 67108864
+  # defaultMaxMemUsage: 10737418240
+  # customMemConfig: TITLE:67108861:1073741824,NAME:67108861:1073741824

scripts/gart

@@ -44,6 +44,10 @@ debug_converter=false
 use_logs_per_epoch=true
 logs_per_epoch=10000
 seconds_per_epoch=60
+default_max_vertex_num=67108864
+default_max_mem_usage=10737418240
+customized_memory_config=
+
 
 set_port=false
 
@@ -128,14 +132,17 @@ Options:
       --use-logs-per-epoch:      Whether to use logs per epoch (default: true).
       --logs-per-epoch:          Number of logs per epoch (default: 10000).
       --seconds-per-epoch:       Number of seconds per epoch (default: 60).
+      --default-max-vertex-num:  Default max vertex number (default: 67108864).
+      --default-max-mem-usage:   Default max memory usage (default: 10737418240).
+      --customized-memory-config: Customized memory config (default: "").
       --debug:                   Specify components to debug. Pass as a comma-separated list.
                                  Examples: '--debug=writer', '--debug=converter,writer' (default: '').
 EOF
 }
 
 ####################### Parse Arguments #######################
 parse_arguments() {
-    local VALID_ARGS=$(getopt -o d:b:u:p:r:t:e:k:h? --long db-host:,db-name:,db-port:,db-user:,db-password:,rgmapping-file:,v6d-sock:,v6d-size:,use-debezium:,db-type:,enable-bulkload:,etcd-endpoint:,etcd-prefix:,kafka-server:,subgraph-num:,subgraph-id:,role:,debug:,start-slot:,num-slot:,k8s-mode:,rg-from-etcd:,use-logs-per-epoch:,logs-per-epoch:,seconds-per-epoch:,help -- "$@")
+    local VALID_ARGS=$(getopt -o d:b:u:p:r:t:e:k:h? --long db-host:,db-name:,db-port:,db-user:,db-password:,rgmapping-file:,v6d-sock:,v6d-size:,use-debezium:,db-type:,enable-bulkload:,etcd-endpoint:,etcd-prefix:,kafka-server:,subgraph-num:,subgraph-id:,role:,debug:,start-slot:,num-slot:,k8s-mode:,rg-from-etcd:,use-logs-per-epoch:,logs-per-epoch:,default-max-vertex-num:,default-max-mem-usage:,customized-memory-config:,seconds-per-epoch:,help -- "$@")
     if [[ $? -ne 0 ]]; then
         exit 1;
     fi
@@ -246,6 +253,18 @@ parse_arguments() {
                 seconds_per_epoch=$2
                 shift 2
                 ;;
+                --default-max-vertex-num)
+                default_max_vertex_num=$2
+                shift 2
+                ;;
+                --default-max-mem-usage)
+                default_max_mem_usage=$2
+                shift 2
+                ;;
+                --customized-memory-config)
+                customized_memory_config=$2
+                shift 2
+                ;;
             -h | --help | ?)
                 show_help
                 exit 0
@@ -713,7 +732,13 @@ if [ "$role" == "writer" ] && [ "$k8s_mode" == "true" ]; then
     vegito_cmd="$WRITER_HOME/vegito --v6d_ipc_socket $v6d_sock \
         --etcd_endpoint $etcd_endpoint --meta_prefix $etcd_prefix \
         --kafka_broker_list $kafka_server \
-        --subgraph_num $subgraph_num --subgraph_id $subgraph_id"
+        --subgraph_num $subgraph_num --subgraph_id $subgraph_id \
+        --default_max_vertex_number $default_max_vertex_num \
+        --default_max_memory_usage_for_each_type_vertex $default_max_mem_usage"
+
+    if [ -n "$customized_memory_config" ]; then
+        vegito_cmd="$vegito_cmd --customized_vertex_number_memory_usage_config $customized_memory_config"
+    fi
 
     if $debug_writer; then
         echo "Debug Wrtier $subgraph_id: ${vegito_cmd} &"

vegito/src/framework/bench_runner.cc

@@ -15,6 +15,7 @@
 #include <stdio.h>
 #include <unistd.h>
 
+#include <cstdint>
 #include <fstream>
 #include <string>
 #include <utility>
@@ -199,6 +200,35 @@ Status init_graph_schema(string etcd_endpoint, string etcd_prefix,
 
   graph_store->init_edge_bitmap_size(elabel_num);
 
+  for (auto idx = 0; idx < vlabel_num; ++idx) {
+    graph_store->set_max_vertex_num(idx, FLAGS_default_max_vertex_number);
+    graph_store->set_max_memory_usage(
+        idx, FLAGS_default_max_memory_usage_for_each_type_vertex);
+  }
+
+  std::cout << "Customized vertex number memory usage config: "
+            << FLAGS_customized_vertex_number_memory_usage_config << std::endl;
+
+  if (!FLAGS_customized_vertex_number_memory_usage_config.empty()) {
+    string customized_config =
+        FLAGS_customized_vertex_number_memory_usage_config;
+    vector<string_view> type_configs = splitString(customized_config, ',');
+    for (auto& type_config : type_configs) {
+      vector<string_view> type_config_parts = splitString(type_config, ':');
+      if (type_config_parts.size() != 3) {
+        LOG(ERROR) << "Invalid customized vertex number memory usage config: "
+                   << type_config;
+        return Status::Invalid();
+      }
+      string type_name(type_config_parts[0]);
+      uint64_t vertex_num = std::stoull(string(type_config_parts[1]));
+      uint64_t memory_usage = std::stoull(string(type_config_parts[2]));
+      int type_id = vertex_name_id_map.find(type_name)->second;
+      graph_store->set_max_vertex_num(type_id, vertex_num);
+      graph_store->set_max_memory_usage(type_id, memory_usage);
+    }
+  }
+
   for (auto idx = 0; idx < vlabel_num; ++idx) {
     graph_store->add_vgraph(idx, rg_map);
   }
@@ -269,9 +299,10 @@ Status init_graph_schema(string etcd_endpoint, string etcd_prefix,
 
       for (int col_idx = 0; col_idx < required_table_schema.size(); ++col_idx) {
         if (required_table_schema[col_idx].size() != 2) {
-          LOG(ERROR) << "Table schema format error. " << "Table name"
-                     << table_name << "Column index " << col_idx << " has "
-                     << required_table_schema[col_idx].size() << " columns.";
+          LOG(ERROR) << "Table schema format error. "
+                     << "Table name" << table_name << "Column index " << col_idx
+                     << " has " << required_table_schema[col_idx].size()
+                     << " columns.";
           assert(false);
         }
         if (required_table_schema[col_idx][0].get<string>() ==
@@ -317,9 +348,9 @@ Status init_graph_schema(string etcd_endpoint, string etcd_prefix,
       }
 
       if (prop_dtype == "") {
-        LOG(ERROR) << "Table schema format error. " << "Table name "
-                   << table_name << " Column name " << prop_table_col_name
-                   << " not found.";
+        LOG(ERROR) << "Table schema format error. "
+                   << "Table name " << table_name << " Column name "
+                   << prop_table_col_name << " not found.";
         assert(false);
       }
 
@@ -544,7 +575,8 @@ void Runner::apply_log_to_store_(const string_view& log, int p_id) {
   }
 
   sv_vec.erase(sv_vec.begin(), sv_vec.begin() + 1);
-  // remove the last element of sv_vec, since the last element of sv_vec is offset of binlog
+  // remove the last element of sv_vec, since the last element of sv_vec is
+  // offset of binlog
   sv_vec.pop_back();
 
   if (op == "add_vertex") {

vegito/src/graph/graph_store.cc

@@ -253,7 +253,10 @@ void GraphStore::add_vgraph(uint64_t vlabel, RGMapping* rg_map) {
   ov_seg_graphs_[vlabel] =
       new seggraph::SegGraph(rg_map, vlabel, block_size, num_vertex);
 #else
-  seg_graphs_[vlabel] = new seggraph::SegGraph(rg_map, vlabel);
+  uint64_t block_size = get_max_memory_usage(vlabel);
+  uint64_t num_vertex = get_max_vertex_num(vlabel);
+  seg_graphs_[vlabel] =
+      new seggraph::SegGraph(rg_map, vlabel, block_size, num_vertex);
 
   // add outer CSR and its schema
   ov_seg_graphs_[vlabel] = new seggraph::SegGraph(rg_map, vlabel);

vegito/src/graph/graph_store.h

@@ -478,6 +478,22 @@ class GraphStore {
 
   void init_external_id_storage(uint64_t vlabel);
 
+  void set_max_vertex_num(uint64_t vlabel, uint64_t num) {
+    max_vertex_num_[vlabel] = num;
+  }
+
+  uint64_t get_max_vertex_num(uint64_t vlabel) {
+    return max_vertex_num_[vlabel];
+  }
+
+  void set_max_memory_usage(uint64_t vlabel, uint64_t usage) {
+    max_memory_usage_[vlabel] = usage;
+  }
+
+  uint64_t get_max_memory_usage(uint64_t vlabel) {
+    return max_memory_usage_[vlabel];
+  }
+
  public:
   IdParser<seggraph::vertex_t> id_parser;
 
@@ -574,6 +590,11 @@ class GraphStore {
   // (vlabel, version) -> vertex property storage snapshot
   std::map<std::pair<uint64_t, uint64_t>, property::Property*>
       property_stores_snapshots_;
+
+  // (vlabel) -> max_vertex_num
+  std::map<uint64_t, uint64_t> max_vertex_num_;
+  // (vlabel) -> max_memory_usage
+  std::map<uint64_t, uint64_t> max_memory_usage_;
 };
 
 }  // namespace graph

vegito/src/system_flags.cc

@@ -29,3 +29,11 @@ DEFINE_string(v6d_ipc_socket, "/var/run/vineyard.sock", "Vineyard IPC socket.");
 
 DEFINE_int32(subgraph_num, 1, "total subgraph number.");
 DEFINE_int32(subgraph_id, 0, "subgraph id.");
+
+DEFINE_int64(default_max_vertex_number, 1 * (1ul << 26),
+             "default max vertex number.");
+DEFINE_int64(default_max_memory_usage_for_each_type_vertex, 10 * (1ul << 30),
+             "default max memory usage for each type vertex.");  // in bytes
+DEFINE_string(customized_vertex_number_memory_usage_config,
+              "",  // format: "type1:100:10000,type2:100:10000"
+              "customized vertex number memory usage config.");

vegito/src/system_flags.h

@@ -29,4 +29,10 @@ DECLARE_string(v6d_ipc_socket);
 DECLARE_int32(subgraph_num);
 DECLARE_int32(subgraph_id);
 
+DECLARE_int64(default_max_vertex_number);
+DECLARE_int64(default_max_memory_usage_for_each_type_vertex);  // in bytes
+DECLARE_string(
+    customized_vertex_number_memory_usage_config);  // format:
+                                                    // "type1:100:10000,type2:100:10000"
+
 #endif  // VEGITO_SRC_SYSTEM_FLAGS_H_