Star658
442
代码
代码
Issues34
Pull Requests120
Actions
项目讨论
Wiki
分析
Star658
442
已合并
fix: dflow对aclInit重复初始化场景做兼容处理 #4441
lining23666创建于 7 天前
fix: dflow对aclInit重复初始化场景做兼容处理 #4441
已合并
lining23666创建于 7 天前
从lining23666/gefix/dflow-acl-repeat-init合入到cann/gedevelop
共 6 个文件变更+253-14
Mdflow/deployer/executor/engine_daemon.cc+11-7
@@ -42,6 +42,7 @@ const char_t *const kArgsKeyDeviceId = "--device_id";
42const char_t *const kArgsKeyMsgQueueDeviceId = "--msg_queue_device_id";42const char_t *const kArgsKeyMsgQueueDeviceId = "--msg_queue_device_id";
43std::atomic<bool> kLoopFlag(true);43std::atomic<bool> kLoopFlag(true);
44std::atomic<bool> acl_initialized{false};44std::atomic<bool> acl_initialized{false};
45+std::atomic<bool> acl_owned_by_dflow{false};
45} // namespace46} // namespace
46 47 
47EngineDaemon::EngineDaemon(bool is_host_cpu) : is_host_cpu_(is_host_cpu) {}48EngineDaemon::EngineDaemon(bool is_host_cpu) : is_host_cpu_(is_host_cpu) {}
@@ -77,12 +78,14 @@ Status EngineDaemon::InitializeWithArgs(int32_t argc, char_t **argv) {
77 MemoryStatisticManager::Instance().Initialize(mem_group_name_);78 MemoryStatisticManager::Instance().Initialize(mem_group_name_);
78 if (!acl_initialized) {79 if (!acl_initialized) {
79 aclError ret = aclInit(nullptr);80 aclError ret = aclInit(nullptr);
80- if (ret != ACL_SUCCESS) {81+ if (ret != ACL_SUCCESS && ret != ACL_ERROR_REPEAT_INITIALIZE) {
81- GELOGE(FAILED, "ACL init failed.");82+ GELOGE(FAILED, "ACL init failed, ret = %d.", static_cast<int32_t>(ret));
82 return FAILED;83 return FAILED;
83- } else {84+ }
84- GELOGI("ACL init success.");85+ GELOGI("ACL init success.");
85- acl_initialized.store(true);86+ acl_initialized.store(true);
87+ if (ret == ACL_SUCCESS) {
88+ acl_owned_by_dflow.store(true);
86 }89 }
87 }90 }
88 return SUCCESS;91 return SUCCESS;
@@ -116,10 +119,11 @@ Status EngineDaemon::InitializeExecutor() {
116}119}
117 120 
118void EngineDaemon::Finalize() {121void EngineDaemon::Finalize() {
119- if (acl_initialized) {122+ if (acl_owned_by_dflow) {
120 aclFinalize();123 aclFinalize();
121- acl_initialized.store(false);124+ acl_owned_by_dflow.store(false);
122 }125 }
126+ acl_initialized.store(false);
123 MemoryStatisticManager::Instance().Finalize();127 MemoryStatisticManager::Instance().Finalize();
124 (void)FinalizeMaintenance();128 (void)FinalizeMaintenance();
125 (void)ge_executor_.Finalize();129 (void)ge_executor_.Finalize();
Mdflow/runner/session/dflow_api.cc+11-7
@@ -24,6 +24,7 @@ namespace {
24constexpr uint32_t kExternalErrorCodeMaxValue = 9999999U; // user define error code max value24constexpr uint32_t kExternalErrorCodeMaxValue = 9999999U; // user define error code max value
25constexpr uint64_t INVALID_SESSION_ID = 0xFFFFFFFFFFFFFFFFULL;25constexpr uint64_t INVALID_SESSION_ID = 0xFFFFFFFFFFFFFFFFULL;
26std::atomic<bool> acl_initialized{false};26std::atomic<bool> acl_initialized{false};
27+std::atomic<bool> acl_owned_by_dflow{false};
27 28 
28void ConvertAscendStringMap(const std::map<ge::AscendString, ge::AscendString> &options,29void ConvertAscendStringMap(const std::map<ge::AscendString, ge::AscendString> &options,
29 std::map<std::string, std::string> &str_options) {30 std::map<std::string, std::string> &str_options) {
@@ -51,12 +52,14 @@ Status DFlowInitialize(const std::map<AscendString, AscendString> &options) {
51 }52 }
52 if (!acl_initialized) {53 if (!acl_initialized) {
53 aclError ret = aclInit(nullptr);54 aclError ret = aclInit(nullptr);
54- if (ret != ACL_SUCCESS) {55+ if (ret != ACL_SUCCESS && ret != ACL_ERROR_REPEAT_INITIALIZE) {
55- GELOGE(FAILED, "ACL init failed.");56+ GELOGE(FAILED, "ACL init failed, ret = %d.", static_cast<int32_t>(ret));
56 return FAILED;57 return FAILED;
57- } else {58+ }
58- GELOGI("ACL init success.");59+ GELOGI("ACL init success.");
59- acl_initialized.store(true);60+ acl_initialized.store(true);
61+ if (ret == ACL_SUCCESS) {
62+ acl_owned_by_dflow.store(true);
60 }63 }
61 }64 }
62 // todo call GEInitialize in new so65 // todo call GEInitialize in new so
@@ -86,10 +89,11 @@ Status DFlowFinalize() {
86 GELOGW("[FINAL]DFlowFinalize is called before DFlowInitialize");89 GELOGW("[FINAL]DFlowFinalize is called before DFlowInitialize");
87 return SUCCESS;90 return SUCCESS;
88 }91 }
89- if (acl_initialized) {92+ if (acl_owned_by_dflow) {
90 aclFinalize();93 aclFinalize();
91- acl_initialized.store(false);94+ acl_owned_by_dflow.store(false);
92 }95 }
96+ acl_initialized.store(false);
93 std::lock_guard<std::mutex> lock(g_dflow_ge_release_mutex);97 std::lock_guard<std::mutex> lock(g_dflow_ge_release_mutex);
94 GELOGT(TRACE_INIT, "DFlowFinalize start.");98 GELOGT(TRACE_INIT, "DFlowFinalize start.");
95 99 
Mtests/dflow/runner/st/testcase/data_flow_session/test_data_flow_api.cc+46-0
@@ -13,6 +13,7 @@
13#include <fstream>13#include <fstream>
14#include "nlohmann/json.hpp"14#include "nlohmann/json.hpp"
15#include "depends/mmpa/src/mmpa_stub.h"15#include "depends/mmpa/src/mmpa_stub.h"
16+#include "depends/ascendcl/src/ascendcl_stub.h"
16#include "utils/mock_execution_runtime.h"17#include "utils/mock_execution_runtime.h"
17 18 
18#include "dflow/runner/session/dflow_api.h"19#include "dflow/runner/session/dflow_api.h"
@@ -120,6 +121,30 @@ ge::dflow::FlowGraph BuildFlowGraph() {
120 flow_graph.SetInputs(inputsOperator).SetOutputs(outputsOperator);121 flow_graph.SetInputs(inputsOperator).SetOutputs(outputsOperator);
121 return flow_graph;122 return flow_graph;
122}123}
124+ 
125+class MockAclApiStubRepeatInit : public AclApiStub {
126+ public:
127+ bool acl_finalize_called = false;
128+ aclError aclInit(const char *configPath) override {
129+ return ACL_ERROR_REPEAT_INITIALIZE;
130+ }
131+ aclError aclFinalize() override {
132+ acl_finalize_called = true;
133+ return ACL_SUCCESS;
134+ }
135+};
136+ 
137+class MockAclApiStubFail : public AclApiStub {
138+ public:
139+ bool acl_finalize_called = false;
140+ aclError aclInit(const char *configPath) override {
141+ return ACL_ERROR_INVALID_PARAM;
142+ }
143+ aclError aclFinalize() override {
144+ acl_finalize_called = true;
145+ return ACL_SUCCESS;
146+ }
147+};
123} // namespace148} // namespace
124class DataFlowApiTest : public testing::Test {149class DataFlowApiTest : public testing::Test {
125 protected:150 protected:
@@ -301,5 +326,26 @@ TEST_F(DataFlowApiTest, FeedRawData) {
301 EXPECT_NE(session3.FeedRawData(graph_id, {raw_data}, 0, data_flow_info, 0), SUCCESS);326 EXPECT_NE(session3.FeedRawData(graph_id, {raw_data}, 0, data_flow_info, 0), SUCCESS);
302 EXPECT_EQ(DFlowFinalize(), SUCCESS);327 EXPECT_EQ(DFlowFinalize(), SUCCESS);
303}328}
329+ 
330+TEST_F(DataFlowApiTest, DFlowInitialize_acl_repeat_init) {
331+ DFlowFinalize();
332+ auto mock_acl = std::make_shared<MockAclApiStubRepeatInit>();
333+ AclApiStub::SetInstance(mock_acl);
334+ std::map<AscendString, AscendString> options = {};
335+ EXPECT_EQ(DFlowInitialize(options), SUCCESS);
336+ EXPECT_EQ(DFlowFinalize(), SUCCESS);
337+ EXPECT_FALSE(mock_acl->acl_finalize_called);
338+ AclApiStub::Reset();
339+}
340+ 
341+TEST_F(DataFlowApiTest, DFlowInitialize_acl_init_failed) {
342+ DFlowFinalize();
343+ auto mock_acl = std::make_shared<MockAclApiStubFail>();
344+ AclApiStub::SetInstance(mock_acl);
345+ std::map<AscendString, AscendString> options = {};
346+ EXPECT_EQ(DFlowInitialize(options), FAILED);
347+ EXPECT_FALSE(mock_acl->acl_finalize_called);
348+ AclApiStub::Reset();
349+}
304} // namespace dflow350} // namespace dflow
305} // namespace ge351} // namespace ge
Mtests/dflow/runner/st/testcase/test_helper_runtime.cc+68-0
@@ -1177,6 +1177,30 @@ ComputeGraphPtr BuildTwoInputDynamicRootGraph(const std::vector<int64_t> &shape,
1177 root_graph->TopologicalSorting();1177 root_graph->TopologicalSorting();
1178 return root_graph;1178 return root_graph;
1179}1179}
1180+ 
1181+class MockAclApiStubRepeatInit : public AclApiStub {
1182+ public:
1183+ bool acl_finalize_called = false;
1184+ aclError aclInit(const char *configPath) override {
1185+ return ACL_ERROR_REPEAT_INITIALIZE;
1186+ }
1187+ aclError aclFinalize() override {
1188+ acl_finalize_called = true;
1189+ return ACL_SUCCESS;
1190+ }
1191+};
1192+ 
1193+class MockAclApiStubFail : public AclApiStub {
1194+ public:
1195+ bool acl_finalize_called = false;
1196+ aclError aclInit(const char *configPath) override {
1197+ return ACL_ERROR_INVALID_PARAM;
1198+ }
1199+ aclError aclFinalize() override {
1200+ acl_finalize_called = true;
1201+ return ACL_SUCCESS;
1202+ }
1203+};
1180} // namespace1204} // namespace
1181 1205 
1182static void StartServer(ge::GrpcServer &grpc_server) {1206static void StartServer(ge::GrpcServer &grpc_server) {
@@ -5807,4 +5831,48 @@ TEST_F(STEST_helper_runtime, TestDeployerDaemonCLient_ProcessMessage) {
5807 RuntimeStub::Reset();5831 RuntimeStub::Reset();
5808 MmpaStub::GetInstance().Reset();5832 MmpaStub::GetInstance().Reset();
5809}5833}
5834+ 
5835+TEST_F(STEST_helper_runtime, TestEngineDaemonAclRepeatInit) {
5836+ mock_handle = (void *)0xffffffff;
5837+ MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpaForHeterogeneousRuntime>());
5838+ RuntimeStub::SetInstance(std::make_shared<MockRuntime2>());
5839+ auto mock_acl = std::make_shared<MockAclApiStubRepeatInit>();
5840+ AclApiStub::SetInstance(mock_acl);
5841+ EngineDaemon engine_daemon;
5842+ auto device_id = std::to_string(0);
5843+ auto queue_id = std::to_string(0);
5844+ auto event_group_id = std::to_string(1);
5845+ const std::string process_name = "npu_executor";
5846+ const char_t *argv[] = {
5847+ process_name.c_str(), "BufferGroupName", queue_id.c_str(), device_id.c_str(),
5848+ event_group_id.c_str(), "--base_dir=./", "--device_id=0", "--msg_queue_device_id=0",
5849+ };
5850+ EXPECT_EQ(engine_daemon.InitializeWithArgs(8, (char **)argv), SUCCESS);
5851+ engine_daemon.Finalize();
5852+ EXPECT_FALSE(mock_acl->acl_finalize_called);
5853+ AclApiStub::Reset();
5854+ MmpaStub::GetInstance().Reset();
5855+}
5856+ 
5857+TEST_F(STEST_helper_runtime, TestEngineDaemonAclInitFailed) {
5858+ mock_handle = (void *)0xffffffff;
5859+ MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpaForHeterogeneousRuntime>());
5860+ RuntimeStub::SetInstance(std::make_shared<MockRuntime2>());
5861+ auto mock_acl = std::make_shared<MockAclApiStubFail>();
5862+ AclApiStub::SetInstance(mock_acl);
5863+ EngineDaemon engine_daemon;
5864+ auto device_id = std::to_string(0);
5865+ auto queue_id = std::to_string(0);
5866+ auto event_group_id = std::to_string(1);
5867+ const std::string process_name = "npu_executor";
5868+ const char_t *argv[] = {
5869+ process_name.c_str(), "BufferGroupName", queue_id.c_str(), device_id.c_str(),
5870+ event_group_id.c_str(), "--base_dir=./", "--device_id=0", "--msg_queue_device_id=0",
5871+ };
5872+ EXPECT_EQ(engine_daemon.InitializeWithArgs(8, (char **)argv), FAILED);
5873+ engine_daemon.Finalize();
5874+ EXPECT_FALSE(mock_acl->acl_finalize_called);
5875+ AclApiStub::Reset();
5876+ MmpaStub::GetInstance().Reset();
5877+}
5810} // namespace ge5878} // namespace ge
Mtests/dflow/runner/ut/dflow/session/dflow_api_unittest.cc+50-0
@@ -13,6 +13,7 @@
13#include "dflow/runner/session/dflow_api.h"13#include "dflow/runner/session/dflow_api.h"
14#include "graph/operator_factory_impl.h"14#include "graph/operator_factory_impl.h"
15#include "depends/mmpa/src/mmpa_stub.h"15#include "depends/mmpa/src/mmpa_stub.h"
16+#include "depends/ascendcl/src/ascendcl_stub.h"
16#include "common/ge_common/ge_types.h"17#include "common/ge_common/ge_types.h"
17#include "graph/ge_local_context.h"18#include "graph/ge_local_context.h"
18#include "graph/ge_global_options.h"19#include "graph/ge_global_options.h"
@@ -224,6 +225,30 @@ class MockMmpa : public ge::MmpaStubApiGe {
224 }225 }
225};226};
226 227 
228+class MockAclApiStubRepeatInit : public AclApiStub {
229+ public:
230+ bool acl_finalize_called = false;
231+ aclError aclInit(const char *configPath) override {
232+ return ACL_ERROR_REPEAT_INITIALIZE;
233+ }
234+ aclError aclFinalize() override {
235+ acl_finalize_called = true;
236+ return ACL_SUCCESS;
237+ }
238+};
239+ 
240+class MockAclApiStubFail : public AclApiStub {
241+ public:
242+ bool acl_finalize_called = false;
243+ aclError aclInit(const char *configPath) override {
244+ return ACL_ERROR_INVALID_PARAM;
245+ }
246+ aclError aclFinalize() override {
247+ acl_finalize_called = true;
248+ return ACL_SUCCESS;
249+ }
250+};
251+ 
227class UtestDflowApi : public testing::Test {252class UtestDflowApi : public testing::Test {
228 protected:253 protected:
229 static void SetUpTestSuite() {254 static void SetUpTestSuite() {
@@ -302,6 +327,31 @@ TEST_F(UtestDflowApi, DFlowInitialize) {
302 ge::MmpaStub::GetInstance().Reset();327 ge::MmpaStub::GetInstance().Reset();
303}328}
304 329 
330+TEST_F(UtestDflowApi, DFlowInitialize_acl_repeat_init) {
331+ EXPECT_EQ(DFlowFinalize(), SUCCESS);
332+ auto mock_acl = std::make_shared<MockAclApiStubRepeatInit>();
333+ AclApiStub::SetInstance(mock_acl);
334+ std::map<AscendString, AscendString> options = {};
335+ ge::MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpa>());
336+ EXPECT_EQ(DFlowInitialize(options), SUCCESS);
337+ EXPECT_EQ(DFlowFinalize(), SUCCESS);
338+ EXPECT_FALSE(mock_acl->acl_finalize_called);
339+ AclApiStub::Reset();
340+ ge::MmpaStub::GetInstance().Reset();
341+}
342+ 
343+TEST_F(UtestDflowApi, DFlowInitialize_acl_init_failed) {
344+ EXPECT_EQ(DFlowFinalize(), SUCCESS);
345+ auto mock_acl = std::make_shared<MockAclApiStubFail>();
346+ AclApiStub::SetInstance(mock_acl);
347+ std::map<AscendString, AscendString> options = {};
348+ ge::MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpa>());
349+ EXPECT_EQ(DFlowInitialize(options), FAILED);
350+ EXPECT_FALSE(mock_acl->acl_finalize_called);
351+ AclApiStub::Reset();
352+ ge::MmpaStub::GetInstance().Reset();
353+}
354+ 
305TEST_F(UtestDflowApi, DFlowInitialize_pne_init_failed) {355TEST_F(UtestDflowApi, DFlowInitialize_pne_init_failed) {
306 auto engine = std::make_shared<TestProcessNodeEngine>();356 auto engine = std::make_shared<TestProcessNodeEngine>();
307 EXPECT_NE(engine, nullptr);357 EXPECT_NE(engine, nullptr);
Mtests/dflow/runner/ut/ge/runtime/heterogeneous/executor/engine_daemon_unittest.cc+67-0
@@ -22,6 +22,7 @@
22#include "ge/ge_api.h"22#include "ge/ge_api.h"
23#include "depends/runtime/src/runtime_stub.h"23#include "depends/runtime/src/runtime_stub.h"
24#include "depends/mmpa/src/mmpa_stub.h"24#include "depends/mmpa/src/mmpa_stub.h"
25+#include "depends/ascendcl/src/ascendcl_stub.h"
25#include "common/util/sanitizer_options.h"26#include "common/util/sanitizer_options.h"
26 27 
27using namespace std;28using namespace std;
@@ -252,6 +253,30 @@ class MockRuntime2 : public RuntimeStub {
252 std::vector<void *> mem_bufs_;253 std::vector<void *> mem_bufs_;
253 MbufStub mbuf_stub_;254 MbufStub mbuf_stub_;
254};255};
256+ 
257+class MockAclApiStubRepeatInit : public AclApiStub {
258+ public:
259+ bool acl_finalize_called = false;
260+ aclError aclInit(const char *configPath) override {
261+ return ACL_ERROR_REPEAT_INITIALIZE;
262+ }
263+ aclError aclFinalize() override {
264+ acl_finalize_called = true;
265+ return ACL_SUCCESS;
266+ }
267+};
268+ 
269+class MockAclApiStubFail : public AclApiStub {
270+ public:
271+ bool acl_finalize_called = false;
272+ aclError aclInit(const char *configPath) override {
273+ return ACL_ERROR_INVALID_PARAM;
274+ }
275+ aclError aclFinalize() override {
276+ acl_finalize_called = true;
277+ return ACL_SUCCESS;
278+ }
279+};
255} // namespace280} // namespace
256class EngineDaemonTest : public testing::Test {281class EngineDaemonTest : public testing::Test {
257 protected:282 protected:
@@ -377,4 +402,46 @@ TEST_F(EngineDaemonTest, TestEngineDaemonHost) {
377 RuntimeStub::Reset();402 RuntimeStub::Reset();
378 GetThreadLocalContext() = back_up_context;403 GetThreadLocalContext() = back_up_context;
379}404}
405+ 
406+TEST_F(EngineDaemonTest, TestEngineDaemonAclRepeatInit) {
407+ MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpa>());
408+ RuntimeStub::SetInstance(std::make_shared<MockRuntime2>());
409+ auto mock_acl = std::make_shared<MockAclApiStubRepeatInit>();
410+ AclApiStub::SetInstance(mock_acl);
411+ EngineDaemon engine_daemon;
412+ auto device_id = std::to_string(0);
413+ auto queue_id = std::to_string(0);
414+ auto event_group_id = std::to_string(1);
415+ const std::string process_name = "npu_executor";
416+ const char_t *argv[] = {
417+ process_name.c_str(), "BufferGroupName", queue_id.c_str(), device_id.c_str(),
418+ event_group_id.c_str(), "--base_dir=./", "--device_id=0", "--msg_queue_device_id=0",
419+ };
420+ EXPECT_EQ(engine_daemon.InitializeWithArgs(8, (char **)argv), SUCCESS);
421+ engine_daemon.Finalize();
422+ EXPECT_FALSE(mock_acl->acl_finalize_called);
423+ AclApiStub::Reset();
424+ MmpaStub::GetInstance().Reset();
425+}
426+ 
427+TEST_F(EngineDaemonTest, TestEngineDaemonAclInitFailed) {
428+ MmpaStub::GetInstance().SetImpl(std::make_shared<MockMmpa>());
429+ RuntimeStub::SetInstance(std::make_shared<MockRuntime2>());
430+ auto mock_acl = std::make_shared<MockAclApiStubFail>();
431+ AclApiStub::SetInstance(mock_acl);
432+ EngineDaemon engine_daemon;
433+ auto device_id = std::to_string(0);
434+ auto queue_id = std::to_string(0);
435+ auto event_group_id = std::to_string(1);
436+ const std::string process_name = "npu_executor";
437+ const char_t *argv[] = {
438+ process_name.c_str(), "BufferGroupName", queue_id.c_str(), device_id.c_str(),
439+ event_group_id.c_str(), "--base_dir=./", "--device_id=0", "--msg_queue_device_id=0",
440+ };
441+ EXPECT_EQ(engine_daemon.InitializeWithArgs(8, (char **)argv), FAILED);
442+ engine_daemon.Finalize();
443+ EXPECT_FALSE(mock_acl->acl_finalize_called);
444+ AclApiStub::Reset();
445+ MmpaStub::GetInstance().Reset();
446+}
380} // namespace ge447} // namespace ge