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 "dflow_api.h"
12 : #include <cinttypes>
13 : #include <atomic>
14 : #include <malloc.h>
15 : #include "dflow_session_manager.h"
16 : #include "common/compile_profiling/ge_call_wrapper.h"
17 : #include "framework/runtime/subscriber/global_profiler.h"
18 : #include "common/option_supportion_checker/option_supportion_checker.h"
19 : #include "dflow/base/exec_runtime/execution_runtime.h"
20 : #include "acl/acl.h"
21 : namespace ge {
22 : namespace dflow {
23 : namespace {
24 : constexpr uint32_t kExternalErrorCodeMaxValue = 9999999U; // user define error code max value
25 : constexpr uint64_t INVALID_SESSION_ID = 0xFFFFFFFFFFFFFFFFULL;
26 : std::atomic<bool> acl_initialized{false};
27 : std::atomic<bool> acl_owned_by_dflow{false};
28 :
29 : void ConvertAscendStringMap(const std::map<ge::AscendString, ge::AscendString> &options,
30 : std::map<std::string, std::string> &str_options) {
31 : for (auto &option_item : options) {
32 : if (option_item.first.GetLength() == 0) {
33 : GELOGE(ge::FAILED, "Construct session failed, option key is empty.");
34 : REPORT_INNER_ERR_MSG("E19999", "Construct session failed, option key is empty.");
35 : return;
36 : }
37 : const std::string &key = option_item.first.GetString();
38 : const std::string &val = option_item.second.GetString();
39 : str_options[key] = val;
40 : }
41 : }
42 :
43 : std::atomic_bool g_dflow_ge_initialized{false};
44 : std::mutex g_dflow_ge_release_mutex; // DFlowInitialize, DFlowFinalize and ~DFlowSession use
45 : std::shared_ptr<DFlowSessionManager> g_dflow_session_manager;
46 :
47 99 : void DFlowFinalizeImpl() {
48 99 : GELOGT(TRACE_INIT, "DFlowFinalize start.");
49 99 : if (g_dflow_session_manager != nullptr) {
50 98 : g_dflow_session_manager->Finalize();
51 : }
52 99 : (void)malloc_trim(0);
53 99 : g_dflow_ge_initialized = false;
54 99 : ge::DFlowFinalizeInner();
55 99 : if (acl_owned_by_dflow) {
56 24 : aclFinalize();
57 24 : acl_owned_by_dflow.store(false);
58 : }
59 99 : acl_initialized.store(false);
60 99 : GELOGT(TRACE_STOP, "DFlowFinalize finished");
61 99 : }
62 : } // namespace
63 :
64 : Status DFlowInitialize(const std::map<AscendString, AscendString> &options) {
65 28 : std::lock_guard<std::mutex> lock(g_dflow_ge_release_mutex);
66 : if (g_dflow_ge_initialized) {
67 : GELOGW("DFlowInitialize is called more than once");
68 : return SUCCESS;
69 : }
70 30 : GE_DISMISSABLE_GUARD(rollback, ([]() { DFlowFinalizeImpl(); }));
71 : if (!acl_initialized) {
72 : aclError ret = aclInit(nullptr);
73 : if (ret != ACL_SUCCESS && ret != ACL_ERROR_REPEAT_INITIALIZE) {
74 : GELOGE(FAILED, "ACL init failed, ret = %d.", static_cast<int32_t>(ret));
75 : return FAILED;
76 : }
77 : GELOGI("ACL init success.");
78 : acl_initialized.store(true);
79 : if (ret == ACL_SUCCESS) {
80 : acl_owned_by_dflow.store(true);
81 : }
82 : }
83 : GE_TIMESTAMP_START(DflowInitializeAll);
84 : GELOGI("sessionManager initial.");
85 : GE_TIMESTAMP_START(DflowSessionManagerInitialize);
86 : g_dflow_session_manager = ge::MakeShared<dflow::DFlowSessionManager>();
87 : if (g_dflow_session_manager == nullptr) {
88 : GELOGE(GE_CLI_INIT_FAILED, "[Init][Create]SessionManager failed");
89 : return FAILED;
90 : }
91 : g_dflow_session_manager->Initialize();
92 : GE_TIMESTAMP_EVENT_END(DflowSessionManagerInitialize, "InnerInitialize::DflowSessionManagerInitialize");
93 :
94 : GE_CHK_STATUS_RET(ge::DFlowInitializeInner(options), "Failed to call dflow initialize inner");
95 : g_dflow_ge_initialized = true;
96 22 : GE_DISMISS_GUARD(rollback);
97 : GELOGT(TRACE_STOP, "DFlowInitialize finished");
98 : GE_TIMESTAMP_EVENT_END(DflowInitializeAll, "DflowInitialize::All");
99 : return SUCCESS;
100 : }
101 :
102 : // DFlow finalize, releasing all resources
103 : Status DFlowFinalize() {
104 : GRAPH_PROFILING_REG(gert::GeProfInfoType::kGEFinalize);
105 : std::lock_guard<std::mutex> lock(g_dflow_ge_release_mutex);
106 95 : DFlowFinalizeImpl();
107 : return SUCCESS;
108 : }
109 :
110 : namespace {
111 : void ConstructSession(const std::map<std::string, std::string> &options, SessionPtr &session_impl) {
112 : GELOGT(TRACE_INIT, "DFlowSession Constructor start");
113 : // check init status
114 : if (!g_dflow_ge_initialized) {
115 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "Construct session failed because GEInitialize was not called before.");
116 : REPORT_INNER_ERR_MSG("E19999", "Construct session failed because GEInitialize was not called before.");
117 : return;
118 : }
119 : // call Initialize
120 : if (ge::GEAPICheckSupportedSessionOptions(options) != SUCCESS) {
121 : GELOGW("[Check][Param] Check supported options failed.");
122 : }
123 : uint64_t tmp_session_id = 0UL;
124 : session_impl = g_dflow_session_manager->CreateSession(options, tmp_session_id);
125 : // failed guarder, should call GE_DISMISS_GUARD if success
126 : GE_DISMISSABLE_GUARD(create_failed,
127 : ([tmp_session_id]() { g_dflow_session_manager->DestroySession(tmp_session_id); }));
128 : if (session_impl == nullptr) {
129 : GELOGE(FAILED, "Construct session failed.");
130 : REPORT_INNER_ERR_MSG("E19999", "Construct session failed.");
131 : return;
132 : }
133 : GE_DISMISS_GUARD(create_failed);
134 : GELOGT(TRACE_STOP, "DFlowSession construct finished, session id is %" PRIu64 "", tmp_session_id);
135 : }
136 : } // namespace
137 :
138 : DFlowSession::DFlowSession(const std::map<AscendString, AscendString> &options) {
139 : std::map<std::string, std::string> str_options;
140 : ConvertAscendStringMap(options, str_options);
141 : ConstructSession(str_options, dflow_session_impl_);
142 : }
143 :
144 : DFlowSession::~DFlowSession() {
145 : if (dflow_session_impl_ == nullptr) {
146 : return;
147 : }
148 : GELOGT(TRACE_INIT, "Start to destroy session.");
149 : // 0.check init status
150 : if (!g_dflow_ge_initialized) {
151 : GELOGW("GE is not yet initialized or is finalized.");
152 : return;
153 : }
154 : Status ret = FAILED;
155 : std::lock_guard<std::mutex> lock(g_dflow_ge_release_mutex);
156 : try {
157 : const uint64_t session_id = dflow_session_impl_->GetSessionId();
158 : // call DestroySession
159 : GELOGT(TRACE_RUNNING, "DFlowSession id is %" PRIu64 "", session_id);
160 : ret = g_dflow_session_manager->DestroySession(session_id);
161 : } catch (std::exception &e) {
162 : (void)e;
163 : GELOGE(GE_CLI_SESS_DESTROY_FAILED, "[Destructor][DFlowSession]Failed: an exception occurred");
164 : REPORT_INNER_ERR_MSG("E19999", "Failed to destroy session: an exception occurred");
165 : }
166 :
167 : // check return status, return, update session id if success
168 : if (ret != SUCCESS) {
169 : GELOGE(ret, "[Destructor][DFlowSession]Failed, error code:%u.", ret);
170 : REPORT_INNER_ERR_MSG("E19999", "Destroy session failed, error code:%u.", ret);
171 : }
172 :
173 : GELOGT(TRACE_STOP, "DFlowSession has been successfully destroyed");
174 : }
175 :
176 : Status DFlowSession::AddGraph(uint32_t graph_id, const FlowGraph &graph,
177 : const std::map<AscendString, AscendString> &options) {
178 : if (!g_dflow_ge_initialized) {
179 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Construct][DFlowSession]Failed because GEInitialize was not called before.");
180 : REPORT_INNER_ERR_MSG("E19999", "Creating session failed because GEInitialize was not called before.");
181 : return FAILED;
182 : }
183 : GE_CHECK_NOTNULL(dflow_session_impl_);
184 : const auto &session_id = dflow_session_impl_->GetSessionId();
185 : const std::string graph_name = graph.GetName();
186 : GE_ASSERT_TRUE((!graph_name.empty()), "Add graph failed, get graph name failed.");
187 : GELOGT(TRACE_INIT, "Start to add graph in DFlowSession. graph_id: %u, graph_name: %s, session_id: %" PRIu64 ".",
188 : graph_id, graph_name.c_str(), session_id);
189 :
190 : std::map<std::string, std::string> str_options;
191 : ConvertAscendStringMap(options, str_options);
192 : if (ge::GEAPICheckSupportedGraphOptions(str_options) != SUCCESS) {
193 : GELOGW("[Check][Param] Check supported options failed.");
194 : }
195 : GELOGD("Adding graph to session");
196 : const Status ret = dflow_session_impl_->AddGraph(graph_id, graph, str_options);
197 : GE_CHK_BOOL_RET_STATUS(ret == SUCCESS, FAILED,
198 : "Add graph failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id,
199 : graph_id);
200 :
201 : GELOGI("AddGraph finished in DFlowSession, graph_id: %u, session_id: %" PRIu64 ".", graph_id, session_id);
202 : return SUCCESS;
203 : }
204 :
205 : Status DFlowSession::RemoveGraph(uint32_t graph_id) {
206 : if (!g_dflow_ge_initialized) {
207 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Construct][DFlowSession]Failed because GEInitialize was not called before.");
208 : REPORT_INNER_ERR_MSG("E19999", "Creating session failed because GEInitialize was not called before.");
209 : return FAILED;
210 : }
211 : GE_CHECK_NOTNULL(dflow_session_impl_);
212 : const auto &session_id = dflow_session_impl_->GetSessionId();
213 : GRAPH_PROFILING_REG(gert::GeProfInfoType::kRemoveGraph);
214 : GELOGT(TRACE_INIT, "DFlowSession RemoveGraph start, graph_id: %u", graph_id);
215 :
216 : // call RemoveGraph
217 : const Status ret = dflow_session_impl_->RemoveGraph(graph_id);
218 : GE_CHK_BOOL_RET_STATUS(ret == SUCCESS, FAILED,
219 : "Remove graph failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id,
220 : graph_id);
221 :
222 : GELOGT(TRACE_STOP, "DFlowSession RemoveGraph finished, graph_id: %u, session_id:%" PRIu64 "", graph_id, session_id);
223 : return ret;
224 : }
225 :
226 : Status DFlowSession::BuildGraph(uint32_t graph_id, const std::vector<ge::Tensor> &inputs) {
227 : if (!g_dflow_ge_initialized) {
228 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Construct][DFlowSession]Failed because GEInitialize was not called before.");
229 : REPORT_INNER_ERR_MSG("E19999", "Creating session failed because GEInitialize was not called before.");
230 : return FAILED;
231 : }
232 :
233 : GE_CHECK_NOTNULL(dflow_session_impl_);
234 : const auto &session_id = dflow_session_impl_->GetSessionId();
235 : GRAPH_PROFILING_REG(gert::GeProfInfoType::kBuildGraph);
236 : GELOGT(TRACE_INIT, "start to build graph, session_id: %" PRIu64 ", graph_id: %u, input size %zu", session_id,
237 : graph_id, inputs.size());
238 :
239 : const Status ret = dflow_session_impl_->BuildGraph(graph_id, inputs);
240 : GE_CHK_BOOL_RET_STATUS(ret == SUCCESS, FAILED,
241 : "Build graph failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id,
242 : graph_id);
243 : GELOGD("BuildGraph finished in DFlowSession, graph_id: %u", graph_id);
244 : return SUCCESS;
245 : }
246 :
247 : uint64_t DFlowSession::GetSessionId() const {
248 : if (dflow_session_impl_ != nullptr) {
249 : return dflow_session_impl_->GetSessionId();
250 : }
251 : return INVALID_SESSION_ID;
252 : }
253 :
254 : Status DFlowSession::FeedDataFlowGraph(uint32_t graph_id, const std::vector<Tensor> &inputs, const DataFlowInfo &info,
255 : int32_t timeout) {
256 : return FeedDataFlowGraph(graph_id, {}, inputs, info, timeout);
257 : }
258 :
259 : Status DFlowSession::FeedDataFlowGraph(uint32_t graph_id, const std::vector<uint32_t> &indexes,
260 : const std::vector<Tensor> &inputs, const DataFlowInfo &info, int32_t timeout) {
261 : if (!g_dflow_ge_initialized) {
262 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Feed][Data]Failed because GEInitialize was not called before.");
263 : REPORT_INNER_ERR_MSG("E19999", "Feed data failed because GEInitialize was not called before.");
264 : return FAILED;
265 : }
266 :
267 : GE_CHECK_NOTNULL(dflow_session_impl_);
268 : const auto &session_id = dflow_session_impl_->GetSessionId();
269 :
270 : GELOGI("Feed data flow graph, graph_id: %u, timeout: %d ms", graph_id, timeout);
271 : const Status ret = dflow_session_impl_->FeedDataFlowGraph(graph_id, indexes, inputs, info, timeout);
272 : if (ret != SUCCESS && ret != ACL_ERROR_GE_REDEPLOYING && ret != ACL_ERROR_GE_SUBHEALTHY) {
273 : GELOGE(ret, "[Feed][Data]Failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id, graph_id);
274 : REPORT_INNER_ERR_MSG("E19999", "Feed data flow graph failed , error code:%u, session_id:%" PRIu64 ", graph_id:%u",
275 : ret, session_id, graph_id);
276 : return (ret > kExternalErrorCodeMaxValue) ? FAILED : ret;
277 : }
278 : return ret;
279 : }
280 :
281 : Status DFlowSession::FeedDataFlowGraph(uint32_t graph_id, const std::vector<FlowMsgPtr> &inputs, int32_t timeout) {
282 : return FeedDataFlowGraph(graph_id, {}, inputs, timeout);
283 : }
284 :
285 : Status DFlowSession::FeedDataFlowGraph(uint32_t graph_id, const std::vector<uint32_t> &indexes,
286 : const std::vector<FlowMsgPtr> &inputs, int32_t timeout) {
287 : GE_CHK_BOOL_RET_STATUS(g_dflow_ge_initialized, FAILED,
288 : "[Feed][FlowMsg]Failed because GEInitialize was not called before.");
289 :
290 : GE_CHECK_NOTNULL(dflow_session_impl_);
291 : const auto &session_id = dflow_session_impl_->GetSessionId();
292 :
293 : GELOGI("Feed flow msg, graph_id: %u, timeout: %d ms", graph_id, timeout);
294 : const Status ret = dflow_session_impl_->FeedDataFlowGraph(graph_id, indexes, inputs, timeout);
295 : const auto status = ret > kExternalErrorCodeMaxValue ? FAILED : ret;
296 : GE_CHK_BOOL_RET_STATUS((ret == SUCCESS || ret == ACL_ERROR_GE_REDEPLOYING || ret == ACL_ERROR_GE_SUBHEALTHY), status,
297 : "[Feed][FlowMsg]Failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id,
298 : graph_id);
299 : return ret;
300 : }
301 :
302 : Status DFlowSession::FeedRawData(uint32_t graph_id, const std::vector<RawData> &raw_data_list, uint32_t index,
303 : const DataFlowInfo &info, int32_t timeout) {
304 : if (!g_dflow_ge_initialized) {
305 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Feed][RawData]Failed because GEInitialize was not called before.");
306 : REPORT_INNER_ERR_MSG("E19999", "Feed raw data failed because GEInitialize was not called before.");
307 : return FAILED;
308 : }
309 : GE_CHECK_NOTNULL(dflow_session_impl_);
310 : const auto &session_id = dflow_session_impl_->GetSessionId();
311 :
312 : GELOGI("Feed raw data to data flow graph, graph_id: %u, timeout: %d ms", graph_id, timeout);
313 : const Status ret = dflow_session_impl_->FeedRawData(graph_id, raw_data_list, index, info, timeout);
314 : if (ret != SUCCESS && ret != ACL_ERROR_GE_REDEPLOYING && ret != ACL_ERROR_GE_SUBHEALTHY) {
315 : GELOGE(ret, "[Feed][Data]Failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id, graph_id);
316 : REPORT_INNER_ERR_MSG("E19999", "Feed data flow graph failed , error code:%u, session_id:%" PRIu64 ", graph_id:%u",
317 : ret, session_id, graph_id);
318 : return (ret > kExternalErrorCodeMaxValue) ? FAILED : ret;
319 : }
320 : return ret;
321 : }
322 :
323 : Status DFlowSession::FetchDataFlowGraph(uint32_t graph_id, std::vector<Tensor> &outputs, DataFlowInfo &info,
324 : int32_t timeout) {
325 : return FetchDataFlowGraph(graph_id, {}, outputs, info, timeout);
326 : }
327 :
328 : Status DFlowSession::FetchDataFlowGraph(uint32_t graph_id, const std::vector<uint32_t> &indexes,
329 : std::vector<Tensor> &outputs, DataFlowInfo &info, int32_t timeout) {
330 : if (!g_dflow_ge_initialized) {
331 : GELOGE(GE_CLI_GE_NOT_INITIALIZED, "[Fetch][Data]Failed because GEInitialize was not called before.");
332 : REPORT_INNER_ERR_MSG("E19999", "Fetch data failed because GEInitialize was not called before.");
333 : return FAILED;
334 : }
335 :
336 : GE_CHECK_NOTNULL(dflow_session_impl_);
337 : const auto &session_id = dflow_session_impl_->GetSessionId();
338 :
339 : GELOGI("Fetch data flow graph, graph_id: %u, timeout: %d ms", graph_id, timeout);
340 : Status ret = dflow_session_impl_->FetchDataFlowGraph(graph_id, indexes, outputs, info, timeout);
341 : const bool need_convert_error_code = ((ret == RT_ERROR_TO_GE_STATUS(ACL_ERROR_RT_QUEUE_EMPTY)) && timeout != 0);
342 : ret = need_convert_error_code ? ACL_ERROR_GE_MODEL_EXECUTE_TIMEOUT : ret;
343 : if (ret != SUCCESS && ret != ACL_ERROR_GE_REDEPLOYING && ret != ACL_ERROR_GE_SUBHEALTHY) {
344 : GELOGE(ret, "[Fetch][Data]Failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id, graph_id);
345 : REPORT_INNER_ERR_MSG("E19999", "Fetch data flow graph failed , error code:%u, session_id:%" PRIu64 ", graph_id:%u",
346 : ret, session_id, graph_id);
347 : return (ret > kExternalErrorCodeMaxValue) ? FAILED : ret;
348 : }
349 : return ret;
350 : }
351 :
352 : Status DFlowSession::FetchDataFlowGraph(uint32_t graph_id, std::vector<FlowMsgPtr> &outputs, int32_t timeout) {
353 : return FetchDataFlowGraph(graph_id, {}, outputs, timeout);
354 : }
355 :
356 : Status DFlowSession::FetchDataFlowGraph(uint32_t graph_id, const std::vector<uint32_t> &indexes,
357 : std::vector<FlowMsgPtr> &outputs, int32_t timeout) {
358 : GE_CHK_BOOL_RET_STATUS(g_dflow_ge_initialized, FAILED,
359 : "[Fetch][FlowMsg]Failed because GEInitialize was not called before.");
360 : GE_CHECK_NOTNULL(dflow_session_impl_);
361 : const auto &session_id = dflow_session_impl_->GetSessionId();
362 :
363 : GELOGI("Fetch flow msg, graph_id: %u, timeout: %d ms", graph_id, timeout);
364 : Status ret = dflow_session_impl_->FetchDataFlowGraph(graph_id, indexes, outputs, timeout);
365 : const bool need_convert_error_code = ((ret == RT_ERROR_TO_GE_STATUS(ACL_ERROR_RT_QUEUE_EMPTY)) && timeout != 0);
366 : ret = need_convert_error_code ? ACL_ERROR_GE_MODEL_EXECUTE_TIMEOUT : ret;
367 : const auto status = ret > kExternalErrorCodeMaxValue ? FAILED : ret;
368 : GE_CHK_BOOL_RET_STATUS((ret == SUCCESS || ret == ACL_ERROR_GE_REDEPLOYING || ret == ACL_ERROR_GE_SUBHEALTHY), status,
369 : "[Fetch][FlowMsg]Failed, error code:%u, session_id:%" PRIu64 ", graph_id:%u.", ret, session_id,
370 : graph_id);
371 : return ret;
372 : }
373 : } // namespace dflow
374 : } // namespace ge
|
