profiling storage detail

vesoft-inc · Aug 30, 2021 · f7ca01d · f7ca01d
1 parent f490d4d
commit f7ca01d
Show file tree

Hide file tree

Showing 32 changed files with 204 additions and 71 deletions.
diff --git a/src/graph/executor/StorageAccessExecutor.h b/src/graph/executor/StorageAccessExecutor.h
@@ -133,8 +133,17 @@ class StorageAccessExecutor : public Executor {
     auto &hostLatency = resp.hostLatency();
     for (size_t i = 0; i < hostLatency.size(); ++i) {
       auto &info = hostLatency[i];
+      auto &response = resp.responses()[i];
       stats.emplace(folly::stringPrintf("%s exec/total", std::get<0>(info).toString().c_str()),
                     folly::stringPrintf("%d(us)/%d(us)", std::get<1>(info), std::get<2>(info)));
+      if (response.result_ref()->latency_detail_us_ref().has_value()) {
+        std::string storage_detail = "{";
+        for (auto iter : (*response.result_ref()->latency_detail_us_ref())) {
+          storage_detail += folly::stringPrintf("%s:%d(us),", iter.first.data(), iter.second);
+        }
+        storage_detail += "}";
+        stats.emplace("storage_detail", storage_detail);
+      }
     }
   }
 

diff --git a/src/graph/executor/query/GetNeighborsExecutor.cpp b/src/graph/executor/query/GetNeighborsExecutor.cpp
@@ -77,6 +77,14 @@ folly::Future<Status> GetNeighborsExecutor::execute() {
               folly::stringPrintf("%s exec/total/vertices", std::get<0>(info).toString().c_str()),
               folly::stringPrintf(
                   "%d(us)/%d(us)/%lu,", std::get<1>(info), std::get<2>(info), size));
+          if (result.result.latency_detail_us_ref().has_value()) {
+            std::string storage_detail = "{";
+            for (auto iter : (*result.result.latency_detail_us_ref())) {
+              storage_detail += folly::stringPrintf("%s:%d(us),", iter.first.data(), iter.second);
+            }
+            storage_detail += "}";
+            otherStats_.emplace("storage_detail", storage_detail);
+          }
         }
         return handleResponse(resp);
       });

diff --git a/src/interface/storage.thrift b/src/interface/storage.thrift
@@ -35,6 +35,7 @@ struct ResponseCommon {
     1: required list<PartitionResult>   failed_parts,
     // Query latency from storage service
     2: required i32                     latency_in_us,
+    3: optional map<string,i32>         latency_detail_us,
 }
 
 

diff --git a/src/storage/BaseProcessor.h b/src/storage/BaseProcessor.h
@@ -43,6 +43,9 @@ class BaseProcessor {
     }
 
     this->result_.set_latency_in_us(this->duration_.elapsedInUSec());
+    if (!profile_detail_.empty()) {
+      this->result_.set_latency_detail_us(std::move(profile_detail_));
+    }
     this->result_.set_failed_parts(this->codes_);
     this->resp_.set_result(std::move(this->result_));
     this->promise_.setValue(std::move(this->resp_));
@@ -102,6 +105,14 @@ class BaseProcessor {
                                      const std::vector<Value>& props,
                                      WriteResult& wRet);
 
+  virtual void profile_detail(const std::string& name, int32_t latency) {
+    if (!profile_detail_.count(name)) {
+      profile_detail_[name] = latency;
+    } else {
+      profile_detail_[name] += latency;
+    }
+  }
+
  protected:
   StorageEnv* env_{nullptr};
   const ProcessorCounters* counters_;
@@ -116,6 +127,8 @@ class BaseProcessor {
   int32_t callingNum_{0};
   int32_t spaceVidLen_;
   bool isIntId_;
+  std::map<std::string, int32_t> profile_detail_;
+  std::mutex profile_mut_;
 };
 
 }  // namespace storage

diff --git a/src/storage/StorageFlags.cpp b/src/storage/StorageFlags.cpp
@@ -47,3 +47,4 @@ DEFINE_bool(query_concurrently,
             false,
             "whether to run query of each part concurrently, only lookup and "
             "go are supported");
+DEFINE_bool(profile_storage_detail, false, "Whether to profile storage plan detail");
diff --git a/src/storage/StorageFlags.h b/src/storage/StorageFlags.h
@@ -35,4 +35,6 @@ DECLARE_int32(max_edge_returned_per_vertex);
 
 DECLARE_bool(query_concurrently);
 
+DECLARE_bool(profile_storage_detail);
+
 #endif  // STORAGE_STORAGEFLAGS_H_
diff --git a/src/storage/exec/AggregateNode.h b/src/storage/exec/AggregateNode.h
@@ -35,13 +35,15 @@ struct PropStat {
 template <typename T>
 class AggregateNode : public IterateNode<T> {
  public:
-  using RelNode<T>::execute;
+  using RelNode<T>::doExecute;
 
   AggregateNode(RuntimeContext* context, IterateNode<T>* upstream, EdgeContext* edgeContext)
-      : IterateNode<T>(upstream), context_(context), edgeContext_(edgeContext) {}
+      : IterateNode<T>(upstream), context_(context), edgeContext_(edgeContext) {
+    IterateNode<T>::name_ = "AggregateNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const T& input) override {
-    auto ret = RelNode<T>::execute(partId, input);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const T& input) override {
+    auto ret = RelNode<T>::doExecute(partId, input);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }
@@ -51,8 +53,8 @@ class AggregateNode : public IterateNode<T> {
     return nebula::cpp2::ErrorCode::SUCCEEDED;
   }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId) override {
-    auto ret = RelNode<T>::execute(partId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId) override {
+    auto ret = RelNode<T>::doExecute(partId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/DeDupNode.h b/src/storage/exec/DeDupNode.h
@@ -18,13 +18,15 @@ namespace storage {
 template <typename T>
 class DeDupNode : public IterateNode<T> {
  public:
-  using RelNode<T>::execute;
+  using RelNode<T>::doExecute;
 
   explicit DeDupNode(nebula::DataSet* resultSet, const std::vector<size_t>& pos)
-      : resultSet_(resultSet), pos_(pos) {}
+      : resultSet_(resultSet), pos_(pos) {
+    IterateNode<T>::name_ = "DedupNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId) override {
-    auto ret = RelNode<T>::execute(partId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId) override {
+    auto ret = RelNode<T>::doExecute(partId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/EdgeNode.h b/src/storage/exec/EdgeNode.h
@@ -52,9 +52,12 @@ class EdgeNode : public IterateNode<T> {
     schemas_ = &(schemaIter->second);
     ttl_ = QueryUtils::getEdgeTTLInfo(edgeContext_, std::abs(edgeType_));
     edgeName_ = edgeContext_->edgeNames_[edgeType_];
+    IterateNode<T>::name_ = "EdgeNode";
   }
 
-  EdgeNode(RuntimeContext* context, EdgeContext* ctx) : context_(context), edgeContext_(ctx) {}
+  EdgeNode(RuntimeContext* context, EdgeContext* ctx) : context_(context), edgeContext_(ctx) {
+    IterateNode<T>::name_ = "EdgeNode";
+  }
 
   RuntimeContext* context_;
   EdgeContext* edgeContext_;
@@ -71,15 +74,17 @@ class EdgeNode : public IterateNode<T> {
 // FetchEdgeNode is used to fetch a single edge
 class FetchEdgeNode final : public EdgeNode<cpp2::EdgeKey> {
  public:
-  using RelNode::execute;
+  using RelNode::doExecute;
 
   FetchEdgeNode(RuntimeContext* context,
                 EdgeContext* edgeContext,
                 EdgeType edgeType,
                 const std::vector<PropContext>* props,
                 StorageExpressionContext* expCtx = nullptr,
                 Expression* exp = nullptr)
-      : EdgeNode(context, edgeContext, edgeType, props, expCtx, exp) {}
+      : EdgeNode(context, edgeContext, edgeType, props, expCtx, exp) {
+    name_ = "FetchEdgeNode";
+  }
 
   bool valid() const override { return valid_; }
 
@@ -91,9 +96,9 @@ class FetchEdgeNode final : public EdgeNode<cpp2::EdgeKey> {
 
   RowReader* reader() const override { return reader_.get(); }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const cpp2::EdgeKey& edgeKey) override {
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const cpp2::EdgeKey& edgeKey) override {
     valid_ = false;
-    auto ret = RelNode::execute(partId, edgeKey);
+    auto ret = RelNode::doExecute(partId, edgeKey);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }
@@ -143,14 +148,16 @@ class FetchEdgeNode final : public EdgeNode<cpp2::EdgeKey> {
 // srcId
 class SingleEdgeNode final : public EdgeNode<VertexID> {
  public:
-  using RelNode::execute;
+  using RelNode::doExecute;
   SingleEdgeNode(RuntimeContext* context,
                  EdgeContext* edgeContext,
                  EdgeType edgeType,
                  const std::vector<PropContext>* props,
                  StorageExpressionContext* expCtx = nullptr,
                  Expression* exp = nullptr)
-      : EdgeNode(context, edgeContext, edgeType, props, expCtx, exp) {}
+      : EdgeNode(context, edgeContext, edgeType, props, expCtx, exp) {
+    name_ = "SingleEdgeNode";
+  }
 
   SingleEdgeIterator* iter() { return iter_.get(); }
 
@@ -164,8 +171,8 @@ class SingleEdgeNode final : public EdgeNode<VertexID> {
 
   RowReader* reader() const override { return iter_->reader(); }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const VertexID& vId) override {
-    auto ret = RelNode::execute(partId, vId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const VertexID& vId) override {
+    auto ret = RelNode::doExecute(partId, vId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/FilterNode.h b/src/storage/exec/FilterNode.h
@@ -24,22 +24,24 @@ data, but not both.
 As for GetNeighbors, it will have filter that involves both tag and edge
 expression. In that case, FilterNode has a upstream of HashJoinNode, which will
 keep popping out edge data. All tage data has been put into ExpressionContext
-before FilterNode is executed. By that means, it can check the filter of tag +
+before FilterNode is doExecuted. By that means, it can check the filter of tag +
 edge.
 */
 template <typename T>
 class FilterNode : public IterateNode<T> {
  public:
-  using RelNode<T>::execute;
+  using RelNode<T>::doExecute;
 
   FilterNode(RuntimeContext* context,
              IterateNode<T>* upstream,
              StorageExpressionContext* expCtx = nullptr,
              Expression* exp = nullptr)
-      : IterateNode<T>(upstream), context_(context), expCtx_(expCtx), filterExp_(exp) {}
+      : IterateNode<T>(upstream), context_(context), expCtx_(expCtx), filterExp_(exp) {
+    IterateNode<T>::name_ = "FilterNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const T& vId) override {
-    auto ret = RelNode<T>::execute(partId, vId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const T& vId) override {
+    auto ret = RelNode<T>::doExecute(partId, vId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/GetNeighborsNode.h b/src/storage/exec/GetNeighborsNode.h
@@ -22,7 +22,7 @@ namespace storage {
 // target cell of a row.
 class GetNeighborsNode : public QueryNode<VertexID> {
  public:
-  using RelNode::execute;
+  using RelNode::doExecute;
 
   GetNeighborsNode(RuntimeContext* context,
                    IterateNode<VertexID>* hashJoinNode,
@@ -35,10 +35,12 @@ class GetNeighborsNode : public QueryNode<VertexID> {
         upstream_(upstream),
         edgeContext_(edgeContext),
         resultDataSet_(resultDataSet),
-        limit_(limit) {}
+        limit_(limit) {
+    name_ = "GetNeighborsNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const VertexID& vId) override {
-    auto ret = RelNode::execute(partId, vId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const VertexID& vId) override {
+    auto ret = RelNode::doExecute(partId, vId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/GetPropNode.h b/src/storage/exec/GetPropNode.h
@@ -16,15 +16,17 @@ namespace storage {
 
 class GetTagPropNode : public QueryNode<VertexID> {
  public:
-  using RelNode<VertexID>::execute;
+  using RelNode<VertexID>::doExecute;
 
   explicit GetTagPropNode(RuntimeContext* context,
                           std::vector<TagNode*> tagNodes,
                           nebula::DataSet* resultDataSet)
-      : context_(context), tagNodes_(std::move(tagNodes)), resultDataSet_(resultDataSet) {}
+      : context_(context), tagNodes_(std::move(tagNodes)), resultDataSet_(resultDataSet) {
+    name_ = "GetTagPropNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const VertexID& vId) override {
-    auto ret = RelNode::execute(partId, vId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const VertexID& vId) override {
+    auto ret = RelNode::doExecute(partId, vId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }
@@ -80,15 +82,17 @@ class GetTagPropNode : public QueryNode<VertexID> {
 
 class GetEdgePropNode : public QueryNode<cpp2::EdgeKey> {
  public:
-  using RelNode::execute;
+  using RelNode::doExecute;
 
   GetEdgePropNode(RuntimeContext* context,
                   std::vector<EdgeNode<cpp2::EdgeKey>*> edgeNodes,
                   nebula::DataSet* resultDataSet)
-      : context_(context), edgeNodes_(std::move(edgeNodes)), resultDataSet_(resultDataSet) {}
+      : context_(context), edgeNodes_(std::move(edgeNodes)), resultDataSet_(resultDataSet) {
+    QueryNode::name_ = "GetEdgePropNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const cpp2::EdgeKey& edgeKey) override {
-    auto ret = RelNode::execute(partId, edgeKey);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const cpp2::EdgeKey& edgeKey) override {
+    auto ret = RelNode::doExecute(partId, edgeKey);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/HashJoinNode.h b/src/storage/exec/HashJoinNode.h
@@ -23,7 +23,7 @@ namespace storage {
 // return a iterator of edges which can pass ttl check and ready to be read.
 class HashJoinNode : public IterateNode<VertexID> {
  public:
-  using RelNode::execute;
+  using RelNode::doExecute;
 
   HashJoinNode(RuntimeContext* context,
                const std::vector<TagNode*>& tagNodes,
@@ -38,10 +38,11 @@ class HashJoinNode : public IterateNode<VertexID> {
         edgeContext_(edgeContext),
         expCtx_(expCtx) {
     UNUSED(tagContext_);
+    IterateNode::name_ = "HashJoinNode";
   }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId, const VertexID& vId) override {
-    auto ret = RelNode::execute(partId, vId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId, const VertexID& vId) override {
+    auto ret = RelNode::doExecute(partId, vId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/IndexEdgeNode.h b/src/storage/exec/IndexEdgeNode.h
@@ -16,7 +16,7 @@ namespace storage {
 template <typename T>
 class IndexEdgeNode final : public RelNode<T> {
  public:
-  using RelNode<T>::execute;
+  using RelNode<T>::doExecute;
 
   IndexEdgeNode(RuntimeContext* context,
                 IndexScanNode<T>* indexScanNode,
@@ -25,10 +25,12 @@ class IndexEdgeNode final : public RelNode<T> {
       : context_(context),
         indexScanNode_(indexScanNode),
         schemas_(schemas),
-        schemaName_(schemaName) {}
+        schemaName_(schemaName) {
+    RelNode<T>::name_ = "IndexEdgeNode";
+  }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId) override {
-    auto ret = RelNode<T>::execute(partId);
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId) override {
+    auto ret = RelNode<T>::doExecute(partId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }

diff --git a/src/storage/exec/IndexFilterNode.h b/src/storage/exec/IndexFilterNode.h
@@ -20,7 +20,7 @@ namespace storage {
 template <typename T>
 class IndexFilterNode final : public RelNode<T> {
  public:
-  using RelNode<T>::execute;
+  using RelNode<T>::doExecute;
 
   // evalExprByIndex_ is true, all fileds in filter is in index. No need to read
   // data anymore.
@@ -30,6 +30,7 @@ class IndexFilterNode final : public RelNode<T> {
                   bool isEdge = false)
       : indexScanNode_(indexScanNode), exprCtx_(exprCtx), filterExp_(exp), isEdge_(isEdge) {
     evalExprByIndex_ = true;
+    RelNode<T>::name_ = "IndexFilterNode";
   }
 
   // evalExprByIndex_ is false, some fileds in filter is out of index, which
@@ -52,9 +53,9 @@ class IndexFilterNode final : public RelNode<T> {
     isEdge_ = false;
   }
 
-  nebula::cpp2::ErrorCode execute(PartitionID partId) override {
+  nebula::cpp2::ErrorCode doExecute(PartitionID partId) override {
     data_.clear();
-    auto ret = RelNode<T>::execute(partId);
+    auto ret = RelNode<T>::doExecute(partId);
     if (ret != nebula::cpp2::ErrorCode::SUCCEEDED) {
       return ret;
     }
Original file line number	Diff line number	Diff line change
Expand Up		@@ -35,4 +35,6 @@ DECLARE_int32(max_edge_returned_per_vertex);

		DECLARE_bool(query_concurrently);

		DECLARE_bool(profile_storage_detail);

		#endif // STORAGE_STORAGEFLAGS_H_