ADD file via upload

graph_runner.cc

@@ -0,0 +1,234 @@
+/**
+ * Copyright 2019 Huawei Technologies Co., Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * Limitations under the License.
+ */
+
+#include "include/transform/graph_ir/graph_runner.h"
+#include <algorithm>
+#include <string>
+#include <memory>
+
+#ifndef ENABLE_LITE_ACL
+#include "pybind11/pybind11.h"
+#endif
+#include "utils/log_adapter.h"
+#include "include/common/utils/config_manager.h"
+#include "sys/time.h"
+#include "include/common/utils/utils.h"
+#include "include/common/utils/callbacks.h"
+#ifdef ENABLE_D
+#include "include/common/utils/callbacks_ge.h"
+#endif
+#include "utils/ms_context.h"
+
+#ifndef ENABLE_LITE_ACL
+namespace py = pybind11;
+#endif
+namespace mindspore {
+namespace transform {
+// 该函数用于创建新的 GE（GraphEngine）会话。
+// 参数 'sess_options' 表示会话的选项。
+std::shared_ptr<ge::Session> GraphRunner::NewSession(const SessionOptions &sess_options) {
+#ifdef ENABLE_D
+  std::shared_ptr<ge::Session> ret; // 用于存储创建的 GE 会话的智能指针
+  auto ms_context = MsContext::GetInstance(); // 获取 MindSpore 上下文实例
+  MS_EXCEPTION_IF_NULL(ms_context);  // 检查上下文实例是否为空
+  if (ms_context->backend_policy() == "ge") {  // 检查当前的后端策略是否为 GE
+    ret = std::make_shared<ge::Session>(sess_options); // 创建一个新的 GE 会话，并使用传入的选项 'sess_options'
+    if (ret == nullptr) {   // 如果创建 GE 会话失败，抛出异常并记录错误日志
+      MS_LOG(EXCEPTION) << "Create GE session failed!";
+    }
+    MS_LOG(INFO) << "Create new GE session success!";  // 记录成功创建 GE 会话的日志
+    return ret;  // 返回新创建的 GE 会话的智能指针
+  }
+#endif
+
+  MS_LOG(WARNING) << "no GE client, return nullptr!"; // 如果没有启用 GE 后端，记录警告日志并返回空指针
+  return nullptr;  // 返回空指针，表示没有创建 GE 会话
+}
+
+// 该构造函数用于初始化GraphRunner对象
+GraphRunner::GraphRunner(const GraphRunnerOptions &options)
+    : options_(options), graph_manager_(DfGraphManager::GetInstance()) {
+  // 检查并记录MindSpore的并行策略是否为ONE_DEVICE
+  if (ConfigManager::GetInstance().parallel_strategy() == ParallelStrategy::ONE_DEVICE) {
+    MS_LOG(INFO) << "ME run in ONE_DEVICE strategy mode";
+  }
+
+  if (options.sess_ptr != nullptr) {  // 根据options中传入的sess_ptr判断是否已有现有会话
+    sess_ = options.sess_ptr;
+  } else {              // 若sess_ptr为空，则调用NewSession函数创建新的GE会话
+    sess_ = NewSession(options.options);
+    if (sess_ == nullptr) {
+      MS_LOG(WARNING) << "graph runner sess_ is nullptr!";
+    }
+  }
+
+#ifdef ENABLE_D
+  auto ms_context = MsContext::GetInstance();
+  MS_EXCEPTION_IF_NULL(ms_context);
+  if (ms_context->backend_policy() == "ge") {
+    // register the callback function
+    // 注册回调函数
+    if (sess_->RegisterCallBackFunc(callbacks::kCheckPoint, callbacks::CheckpointSaveCallback) != ge::GRAPH_SUCCESS) {
+      MS_LOG(EXCEPTION) << "register callback failed!";
+    }
+
+    if (sess_->RegisterCallBackFunc(callbacks::kSummary, callbacks::SummarySaveCallback) != ge::GRAPH_SUCCESS) {
+      MS_LOG(EXCEPTION) << "register summary callback failed!";
+    }
+  }
+#endif
+  // 从图形管理器获取所有的图形包装器
+  std::vector<DfGraphWrapperPtr> wrappers = graph_manager_.GetAllGraphs();
+  if (wrappers.empty()) {   // 若图形包装器为空，记录日志并直接返回
+    MS_LOG(INFO) << "The GraphManager is empty!!";
+    return;
+  }
+#ifdef ENABLE_D
+  if (ms_context->backend_policy() != "ge") {
+    return;
+  }
+
+  for (auto &it : wrappers) {   // 遍历所有图形包装器，并将未保存的图形添加到GE会话中
+    std::set<string> saved_graph = graph_manager_.GetSavedGraphs();
+    auto iter_find = saved_graph.find(std::to_string(it->id_));
+    if (iter_find != saved_graph.end()) {
+      continue;
+    }
+    MS_LOG(INFO) << "Add the graph " << (*it).name_ << " to GE, it's id is: " << (*it).id_;
+    graph_manager_.AddSavedGraphs(std::to_string(it->id_));
+    (void)sess_->AddGraph(static_cast<uint32_t>(it->id_), *(it->graph_ptr_), it->options_);
+  }
+#endif
+}
+
+// 该函数用于运行指定名称的图形（Graph）。
+Status GraphRunner::RunGraph(const RunOptions &options, const std::vector<GeTensorPtr> &inputs,
+                             std::vector<GeTensorPtr> *outputs) {
+  std::string name = options.name;  // 获取运行选项中的图形名称
+  if (name.empty()) {      // 如果图形名称为空，记录错误日志并返回无效参数状态
+    MS_LOG(ERROR) << "The graph name is null";
+    return Status::INVALID_ARGUMENT;
+  }
+  // 从图形管理器获取指定名称的图形包装器
+  DfGraphWrapperPtr wrap_ptr = graph_manager_.GetGraphByName(name);
+  if (wrap_ptr == nullptr) {
+    MS_LOG(ERROR) << "Get graph form DfGraphManager failed!";  // 如果获取图形包装器失败，记录未找到状态的错误日志
+    return Status::NOT_FOUND;
+  }
+
+  if (wrap_ptr->graph_ptr_ == nullptr) {  // 如果图形为空，记录警告日志并返回未找到状态
+    MS_LOG(WARNING) << "The graph is null";
+    return Status::NOT_FOUND;
+  }
+
+  // call ge::RunGraph() to exec a graph;
+  // 调用 ge::RunGraph() 来执行图形计算
+  std::vector<GeTensor> ge_inputs;
+  std::vector<GeTensor> ge_outputs;
+
+  // 将输入参数 'inputs' 转换为 'ge_inputs'，用于调用 GE 接口
+  (void)std::transform(inputs.begin(), inputs.end(), std::back_inserter(ge_inputs),
+                       [](const GeTensorPtr &i) { return *i; });
+
+  MS_LOG(INFO) << "Run the graph in GE with " << ge_inputs.size() << " inputs"; // 记录日志，表示正在运行 GE 图形，以及输入的数量
+
+  struct timeval start_time, end_time;
+  (void)gettimeofday(&start_time, nullptr);
+
+#ifdef ENABLE_D
+  auto ms_context = MsContext::GetInstance();
+  MS_EXCEPTION_IF_NULL(ms_context);
+  if (ms_context->backend_policy() == "ge") {
+    if (sess_ == nullptr) {
+      MS_LOG(ERROR) << "The GE session is null, can't run the graph!";  // 如果 GE 会话为空，记录错误日志并返回执行失败状态
+      return Status::FAILED;
+    }
+    ge::Status ret = sess_->RunGraph(static_cast<uint32_t>(wrap_ptr->id_), ge_inputs, ge_outputs); // 调用 GE 接口运行图形
+    if (ret != ge::GRAPH_SUCCESS) {
+      MS_LOG(ERROR) << "Call GE RunGraph Failed, ret is: " << ret;  // 如果运行图形失败，记录错误日志并返回执行失败状态
+      return Status::FAILED;
+    }
+  }
+#else
+  ge_outputs.swap(ge_inputs);  // 如果未启用 GE 后端，直接交换输入和输出，用于后续返回输出结果
+#endif
+
+  (void)gettimeofday(&end_time, nullptr);
+  const uint64_t kUSecondInSecond = 1000000;
+  uint64_t cost = kUSecondInSecond * static_cast<uint64_t>(end_time.tv_sec - start_time.tv_sec);
+  cost += static_cast<uint64_t>(end_time.tv_usec - start_time.tv_usec);
+  MS_LOG(INFO) << "Call GE RunGraph Success in " << cost << " us, the GE outputs num is: " << ge_outputs.size();
+  // 记录日志，表示图形计算成功执行，并打印执行时间和输出数量
+
+    // 将 GE 输出结果转换为 'outputs'，用于返回给调用者
+  (void)std::transform(ge_outputs.begin(), ge_outputs.end(), std::back_inserter(*outputs),
+                       [](const GeTensor &ge_tensor) { return std::make_shared<GeTensor>(ge_tensor); });
+
+  return Status::SUCCESS;  // 返回执行成功状态，并带有输出结果
+}
+
+// 该函数用于运行指定名称的图形，并将输入和输出都转换为 MeTensorPtr 类型
+Status GraphRunner::RunGraph(const RunOptions &options, const std::vector<MeTensorPtr> &inputs,
+                             std::vector<MeTensorPtr> *const outputs) {
+  std::vector<GeTensorPtr> ge_inputs;  // 用于存储转换后的输入 GeTensorPtr
+  for (auto it : inputs) {
+    MS_EXCEPTION_IF_NULL(it);
+    MS_LOG(INFO) << "inputs tensor's data size is: " << (*it).DataSize(); // 打印输入 MeTensor 的数据大小
+    auto shape = (*it).shape();
+    std::string shape_str;
+    for (const auto &elem : shape) {
+      shape_str += std::to_string(elem);
+      shape_str += " ";
+    }
+    MS_LOG(INFO) << "inputs tensor's shape is: { " << shape_str << "}";  // 打印输入 MeTensor 的形状
+
+     // 将输入 MeTensor 转换为 GeTensor，转换后的格式为 kOpFormat_NCHW
+    auto ge_tensor_ptr = TransformUtil::ConvertTensor(it, kOpFormat_NCHW);
+    if (ge_tensor_ptr != nullptr) {
+      ge_inputs.emplace_back(ge_tensor_ptr);  // 将转换后的 GeTensorPtr 添加到 ge_inputs 中
+    } else {         // 如果转换失败，记录日志并返回执行失败状态
+      MS_LOG(INFO) << "Convert input Me tensor to Ge tensor failed. Abort this graph"; 
+      return Status::FAILED;
+    }
+  }
+
+  std::vector<GeTensorPtr> ge_outputs;  // 用于存储运行图形后的输出 GeTensorPtr
+  Status ret;
+  {
+    // Release GIL before calling into (potentially long-running) C++ code
+    // 释放 GIL，然后调用 C++ 代码（可能是长时间运行的代码）
+#ifndef ENABLE_LITE_ACL
+    py::gil_scoped_release release;
+#endif
+    ret = RunGraph(options, ge_inputs, &ge_outputs);  // 调用 RunGraph 函数运行图形计算
+  }
+  if (ret != Status::SUCCESS) {
+    return ret;     // 如果运行图形失败，直接返回执行失败状态
+  } else {
+    // convert GeTensor to MeTensor
+    // 将输出 GeTensor 转换为 MeTensor，并将转换后的 MeTensorPtr 添加到 outputs 中
+    for (auto &it : ge_outputs) {
+      auto tensor = TransformUtil::ConvertGeTensor(it);
+      if (tensor != nullptr) {
+        (void)outputs->emplace_back(tensor);
+      }
+    }
+    MS_LOG(INFO) << "Return Me tensor outputs num is: " << outputs->size();  // 打印返回的 MeTensor 数量
+    return Status::SUCCESS;  // 返回执行成功状态，并带有输出结果
+  }
+}
+}  // namespace transform
+}  // namespace mindspore