Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "graph/manager/graph_manager_utils.h"
12 : #include <set>
13 : #include "common/checker.h"
14 : #include "graph/debug/ge_attr_define.h"
15 : #include "graph/ge_context.h"
16 : #include "ge/ge_api_types.h"
17 : #include "base/err_msg.h"
18 : #include "common/memory/tensor_trans_utils.h"
19 :
20 : namespace ge {
21 : namespace {
22 : const char_t *kFrozenInputIndexes = "ge.exec.frozenInputIndexes";
23 :
24 : constexpr size_t kIndexOfFrozenDataIndex = 0UL;
25 : constexpr size_t kIndexOfFrozenDataAddr = 1UL;
26 : constexpr size_t kIndexOfFrozenDataLen = 2UL;
27 :
28 : Status IsDigitStrings(const std::vector<std::string> &input_vec) {
29 : for (const auto &input : input_vec) {
30 : for (const char ch : input) {
31 : if (ch != ' ' && (static_cast<bool>(isdigit(static_cast<unsigned char>(ch))) == false)) {
32 : (void)REPORT_PREDEFINED_ERR_MSG("E10001", std::vector<const char_t *>({"parameter", "value", "reason"}),
33 : std::vector<const char_t *>({kFrozenInputIndexes, input.c_str(),
34 : "The frozen input index is not a digit."}));
35 : return GE_GRAPH_OPTIONS_INVALID;
36 : }
37 : }
38 : }
39 : return SUCCESS;
40 : }
41 : } // namespace
42 : const char_t *GetRunGraphModeStr(RunGraphMode mode) {
43 : static constexpr const char_t *run_graph_mode_str[] = {"RunGraph", "RunGraphAsync", "RunGraphWithStreamAsync",
44 : "InitRunGraphMode"};
45 :
46 : static_assert(sizeof(run_graph_mode_str) / sizeof(run_graph_mode_str[0]) ==
47 : (static_cast<size_t>(RunGraphMode::kRunGraphModeEnd) + 1U),
48 : "RunGraphMode enum and string array size mismatch");
49 :
50 : const auto index = static_cast<size_t>(mode);
51 : if (index >= static_cast<size_t>(RunGraphMode::kRunGraphModeEnd)) {
52 : return "UnknownRunGraphMode";
53 : }
54 : return run_graph_mode_str[index];
55 : }
56 :
57 : GraphNode::GraphNode(const GraphId graph_id) : graph_id_(graph_id), sem_(1U), const_mem_(), feature_mem_() {
58 : std::string opt;
59 : (void)GetContext().GetOption(GRAPH_MAX_PARALLEL_MODEL_NUM, opt);
60 : int32_t max_num = 0;
61 1407 : if (!opt.empty() && (ConvertToInt32(opt, max_num) == SUCCESS) && max_num > 0) {
62 5 : max_load_record_ = static_cast<uint32_t>(max_num);
63 : }
64 : GELOGD("[GraphManager] graphMaxParallelModelNum is %u", max_load_record_);
65 : }
66 :
67 : GraphNode::~GraphNode() = default;
68 :
69 : void GraphNode::Lock() {
70 : (void)sem_.Push(0U);
71 : }
72 :
73 : void GraphNode::Unlock() {
74 : uint8_t unused;
75 : (void)sem_.Pop(unused);
76 : (void)unused;
77 : }
78 :
79 : void GraphNode::IncreaseLoadCount() {
80 : const std::unique_lock<std::mutex> lock(load_count_mu_);
81 : if (load_record_ == max_load_record_) {
82 : GELOGW("Reach the maximum of load_count:%u", kMaxLoadNum);
83 : return;
84 : }
85 : ++load_count_;
86 : }
87 :
88 : Status GraphNode::ParseFrozenInputIndex() {
89 : std::string frozen_input;
90 : (void)ge::GetContext().GetOption(kFrozenInputIndexes, frozen_input);
91 : if (frozen_input.empty()) {
92 : return SUCCESS;
93 : }
94 : // frozen input: ids(0;1;2) id,addr,len(0,100,4;1,200,4)
95 : std::vector<std::string> frozen_input_vec = StringUtils::Split(frozen_input, ';');
96 : for (auto &frozen_info : frozen_input_vec) {
97 : std::vector<std::string> frozen_info_vec = StringUtils::Split(frozen_info, ',');
98 : GE_ASSERT_TRUE((frozen_info_vec.size() == 1UL) || (frozen_info_vec.size() == 3UL),
99 : "frozen info vector size should be 1 or 3 parsed by [%s]", frozen_info.c_str());
100 : GE_ASSERT_SUCCESS(IsDigitStrings(frozen_info_vec), "There are some invalid characters in frozen option value[%s].",
101 : frozen_input.c_str());
102 : int32_t frozen_input_index = -1;
103 : GE_ASSERT_SUCCESS(ConvertToInt32(frozen_info_vec[kIndexOfFrozenDataIndex], frozen_input_index));
104 : GE_ASSERT_TRUE((frozen_input_index >= 0), "Frozen_input_index must be greater than zero: %u", frozen_input_index);
105 : (void)frozen_input_indexes_.insert(static_cast<uint32_t>(frozen_input_index));
106 : if (frozen_info_vec.size() == 1UL) {
107 : GELOGD("Parse frozen input index[%d] success.", frozen_input_index);
108 : continue;
109 : }
110 : uint64_t addr = 0UL;
111 : GE_ASSERT_SUCCESS(ConvertToUint64(frozen_info_vec[kIndexOfFrozenDataAddr], addr));
112 : GE_ASSERT_TRUE(addr != 0UL, "Frozen input addr cannot be nullptr.");
113 : uint64_t len = 0UL;
114 : GE_ASSERT_SUCCESS(ConvertToUint64(frozen_info_vec[kIndexOfFrozenDataLen], len));
115 : GE_ASSERT_TRUE(len != 0UL, "Frozen input length cannot be zero.");
116 : frozen_index_to_node_info_[static_cast<uint32_t>(frozen_input_index)] = std::make_pair(addr, len);
117 : GELOGI("Parse and set frozen addr[%lx] length[%lu] for input index[%d] success.", addr, len, frozen_input_index);
118 : }
119 : return SUCCESS;
120 : }
121 :
122 : void GraphNode::SetLoaded() {
123 : --load_count_;
124 : ++load_record_;
125 : load_flag_ = true;
126 : }
127 : // 只fork编译时信息
128 : std::shared_ptr<GraphNode> GraphNode::Fork(uint32_t forked_graph_id) {
129 : GraphNodePtr graph_node = MakeShared<GraphNode>(forked_graph_id);
130 : GE_ASSERT_NOTNULL(graph_node, "[Fork][GraphNode] fail, graph_id:%u", forked_graph_id);
131 : graph_node->origin_graph_id_ = this->graph_id_;
132 : graph_node->options_ = this->options_;
133 : graph_node->context_ = this->context_;
134 : graph_node->graph_ = this->graph_;
135 : graph_node->compute_graph_ = this->compute_graph_;
136 : graph_node->compiled_flag_ = this->compiled_flag_;
137 : graph_node->build_flag_ = this->GetBuildFlag();
138 : if (this->GetBuildFlag()) {
139 : // 当前状态机管理不好，build flag意味着编译加载成功
140 : // 因此fork时若原始graph node已经build成功，则fork的node也认为build成功
141 : graph_node->compiled_flag_ = true;
142 : }
143 : graph_node->async_ = this->async_;
144 : graph_node->is_specific_stream_ = this->is_specific_stream_;
145 : // todo GeModelPtr ge_model_;
146 : GE_ASSERT_NOTNULL(ge_root_model_);
147 : auto forked_root_model = ge_root_model_->Fork();
148 : GE_ASSERT_NOTNULL(forked_root_model);
149 : graph_node->ge_root_model_ = forked_root_model;
150 : graph_node->is_feature_base_refreshable_ = this->is_feature_base_refreshable_;
151 : // 如下三个数据结构跨越了编译与加载，
152 : // first是地址属于加载阶段数据，second是size属于编译阶段数据（虽然是在loadgraph接口中生成的） 在当前的实现中，
153 : // loadgraph接口生成compile
154 : // summary，并将如下三个数据结构中的size字段设置好。在加载时分配内存的时候，直接读数据结构中的size字段
155 : // 并且loadgraph接口判断若已存在compile summary，便不再生成summary，也不再设置size
156 : // 这样意味着，如下三个数据结构中的size字段和compile
157 : // summary数据存在冗余，且数据是不同步的。坏味道：编译、加载阶段数据没有解耦
158 : // 对于fork接口，需要全盘接收编译时信息。如下三个size字段也算。
159 : // 临时方案：fork这里先复制
160 : // 正式方案：将如上的数据解耦
161 : graph_node->const_mem_ = std::make_pair(nullptr, this->const_mem_.second);
162 : graph_node->feature_mem_ = std::make_pair(nullptr, this->feature_mem_.second);
163 : graph_node->refreshable_feature_mem_ = std::make_pair(nullptr, this->refreshable_feature_mem_.second);
164 :
165 : graph_node->compiled_summary_ = this->compiled_summary_;
166 : graph_node->net_output_node_ = this->net_output_node_;
167 : graph_node->tensor_sizes_ = this->tensor_sizes_;
168 : graph_node->is_saved_net_output_tensor_info_flag_ = this->is_saved_net_output_tensor_info_flag_;
169 : graph_node->ge_tensor_descs_ = this->ge_tensor_descs_;
170 : graph_node->group_2_communication_nodes_ = this->group_2_communication_nodes_;
171 : return graph_node;
172 : }
173 :
174 : SubGraphInfo::SubGraphInfo() : subgraph_ptr_(nullptr), ge_model_ptr_(nullptr) {}
175 :
176 : SubGraphInfo::~SubGraphInfo() = default;
177 :
178 : GraphModelListener::GraphModelListener() : ModelListener() {}
179 :
180 : Status GraphModelListener::OnComputeDone(const uint32_t model_id, const uint32_t data_index, const uint32_t result_code,
181 : std::vector<gert::Tensor> &outputs) {
182 : (void)outputs;
183 : GELOGI(
184 : "[GraphManager] graph compute call back, model_id:%u, task_id:%u, "
185 : "resultCode:%u.",
186 : model_id, data_index, result_code);
187 :
188 : const std::lock_guard<std::mutex> lock(mutex_);
189 : result_code_ = result_code;
190 : is_finished_ = true;
191 : condition_.notify_all();
192 :
193 : return SUCCESS;
194 : }
195 :
196 : uint32_t GraphModelListener::GetResultCode() {
197 : // Pending until async execute graph complete
198 : std::unique_lock<std::mutex> lock(mutex_);
199 : if (!is_finished_) {
200 : GELOGI("[GetResultCode] wait model execute finished.");
201 : condition_.wait(lock);
202 : }
203 :
204 : if (!is_finished_) {
205 : REPORT_INNER_ERR_MSG("E19999", "Model not run finish");
206 : GELOGE(INTERNAL_ERROR, "[Check][Param] model not run finish.");
207 : return INTERNAL_ERROR;
208 : }
209 : return result_code_;
210 : }
211 :
212 : Status GraphModelListener::ResetResult() {
213 : const std::lock_guard<std::mutex> lock(mutex_);
214 : result_code_ = 0U;
215 : is_finished_ = false;
216 :
217 : return SUCCESS;
218 : }
219 :
220 : void RunAsyncListener::SetCallback(const RunAsyncCallbackV2 &callback) {
221 : (void)sem_v2_.Push(callback);
222 : }
223 :
224 : Status RunAsyncListener::OnComputeDone(const uint32_t model_id, const uint32_t data_index, const uint32_t result_code,
225 : std::vector<gert::Tensor> &outputs) {
226 : GELOGI("[GraphManager] run graph async call back, modelId:%u, taskId:%u, resultCode:%u.", model_id, data_index,
227 : result_code);
228 : RunAsyncCallbackV2 callback;
229 : (void)sem_v2_.Pop(callback, 0U); // pop with no wait
230 : GE_CHECK_NOTNULL(callback);
231 : callback(result_code, outputs);
232 : return SUCCESS;
233 : }
234 : } // namespace ge
|
