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 "aicpu_async_event.h"
12 : #include <map>
13 : #include <mutex>
14 : #include "aicpu_sharder_log.h"
15 : #include "aicpu_context.h"
16 : #include "driver/ascend_hal.h"
17 : #include "driver/ascend_hal_define.h"
18 :
19 : namespace aicpu {
20 : namespace {
21 : struct AsyncEventInfo {
22 : uint32_t eventId;
23 : uint32_t subEventId;
24 :
25 : bool operator == (const AsyncEventInfo &info) const
26 : {
27 : return (eventId == info.eventId) && (subEventId == info.subEventId);
28 : }
29 : friend bool operator < (const AsyncEventInfo &info1, const AsyncEventInfo &info2);
30 : };
31 :
32 51 : inline bool operator < (const AsyncEventInfo &info1, const AsyncEventInfo &info2)
33 : {
34 85 : return (info1.eventId < info2.eventId) ||
35 85 : ((info1.eventId == info2.eventId) && (info1.subEventId < info2.subEventId));
36 : }
37 :
38 : struct AsyncTaskInfo {
39 : uint64_t startTick;
40 : std::string opName;
41 : uint8_t waitType;
42 : uint32_t waitId;
43 : uint64_t taskId;
44 : uint32_t streamId;
45 : int32_t currentTimes;
46 : int32_t maxTimes;
47 : EventProcessCallBack taskCb;
48 : };
49 :
50 : std::mutex g_mapMutex;
51 : std::map<AsyncEventInfo, AsyncTaskInfo> g_asyncTaskMap;
52 :
53 :
54 : struct OpInfo {
55 : uint64_t taskId;
56 : uint32_t streamId;
57 : uint32_t threadIndex;
58 :
59 : bool operator == (const OpInfo &info) const
60 : {
61 : return (taskId == info.taskId) && (streamId == info.streamId) && (threadIndex == info.threadIndex);
62 : }
63 : friend bool operator < (const OpInfo &info1, const OpInfo &info2);
64 : };
65 :
66 30 : inline bool operator < (const OpInfo &info1, const OpInfo &info2)
67 : {
68 30 : if (info1.taskId != info2.taskId) {
69 7 : return info1.taskId < info2.taskId;
70 : }
71 :
72 23 : if (info1.streamId != info2.streamId) {
73 5 : return info1.streamId < info2.streamId;
74 : }
75 :
76 18 : if (info1.threadIndex != info2.threadIndex) {
77 6 : return info1.threadIndex < info2.threadIndex;
78 : }
79 12 : return false;
80 : }
81 : std::mutex g_opMapMutex;
82 : std::map<AsyncEventInfo, std::map<OpInfo, EventProcessCallBack>> g_opAsyncTaskMap;
83 :
84 8 : bool GenTaskInfoFromCtx(AsyncTaskInfo &taskInfo)
85 : {
86 8 : (void)aicpu::GetTaskAndStreamId(taskInfo.taskId, taskInfo.streamId);
87 8 : std::string waitIdValue;
88 8 : auto status = aicpu::GetThreadLocalCtx(aicpu::CONTEXT_KEY_WAIT_ID, waitIdValue);
89 8 : if (status != aicpu::AICPU_ERROR_NONE) {
90 0 : AICPUE_LOGE("GetThreadLocalCtx failed, ret=%d, key=%s.", static_cast<int32_t>(status),
91 : aicpu::CONTEXT_KEY_WAIT_ID.c_str());
92 0 : return false;
93 : }
94 8 : int32_t waitId = 0;
95 : try {
96 8 : waitId = std::stoi(waitIdValue);
97 1 : } catch (...) {
98 1 : AICPUE_LOGE("Transfer string:%s to waitId failed", waitIdValue.c_str());
99 1 : return false;
100 1 : }
101 7 : taskInfo.waitId = static_cast<uint32_t>(waitId);
102 7 : std::string waitTypeValue;
103 7 : status = aicpu::GetThreadLocalCtx(aicpu::CONTEXT_KEY_WAIT_TYPE, waitTypeValue);
104 7 : if (status != aicpu::AICPU_ERROR_NONE) {
105 0 : AICPUE_LOGE("GetThreadLocalCtx failed, ret=%d, key=%s.", static_cast<int32_t>(status),
106 : aicpu::CONTEXT_KEY_WAIT_TYPE.c_str());
107 0 : return false;
108 : }
109 7 : int32_t waitType = 0;
110 : try {
111 7 : waitType = std::stoi(waitTypeValue);
112 1 : } catch (...) {
113 1 : AICPUE_LOGE("Transfer string:%s to waitId failed", waitTypeValue.c_str());
114 1 : return false;
115 1 : }
116 6 : taskInfo.waitType = static_cast<uint8_t>(waitType);
117 6 : if (&aicpu::aicpuGetProfContext != nullptr) {
118 6 : const aicpu::aicpuProfContext_t &aicpuProfCtx = aicpu::aicpuGetProfContext();
119 6 : taskInfo.startTick = aicpuProfCtx.tickBeforeRun;
120 : }
121 6 : status = aicpu::GetOpname(aicpu::GetAicpuThreadIndex(), taskInfo.opName);
122 6 : if (status != aicpu::AICPU_ERROR_NONE) {
123 0 : AICPUE_LOGE("GetOpname failed, ret=%d.", static_cast<int32_t>(status));
124 0 : return false;
125 : }
126 6 : return true;
127 8 : }
128 : }
129 :
130 3 : AsyncEventManager::AsyncEventManager() : notifyFunc_(nullptr) {}
131 :
132 3 : AsyncEventManager::~AsyncEventManager() {}
133 :
134 88 : AsyncEventManager &AsyncEventManager::GetInstance()
135 : {
136 88 : static AsyncEventManager asyncEventMgr;
137 88 : return asyncEventMgr;
138 : }
139 :
140 39 : void AsyncEventManager::Register(const NotifyFunc ¬ify)
141 : {
142 39 : notifyFunc_ = notify;
143 39 : }
144 :
145 7 : void AsyncEventManager::NotifyWait(void * const notifyParam, const uint32_t paramLen)
146 : {
147 7 : if (notifyFunc_ != nullptr) {
148 7 : notifyFunc_(notifyParam, paramLen);
149 : }
150 7 : return;
151 : }
152 :
153 8 : bool AsyncEventManager::RegEventCb(const uint32_t eventId, const uint32_t subEventId, const EventProcessCallBack &cb,
154 : const int32_t times)
155 : {
156 8 : if (cb == nullptr) {
157 0 : AICPUE_LOGE("AsyncEventManager RegEventCb failed, cb is nullptr.");
158 0 : return false;
159 : }
160 8 : AsyncTaskInfo taskInfo;
161 8 : taskInfo.taskCb = cb;
162 8 : taskInfo.currentTimes = 0;
163 8 : taskInfo.maxTimes = times;
164 8 : if (!GenTaskInfoFromCtx(taskInfo)) {
165 2 : AICPUE_LOGE("AsyncEventManager GenTaskInfoFromCtx failed.");
166 2 : return false;
167 : }
168 : AsyncEventInfo info;
169 6 : info.eventId = eventId;
170 6 : info.subEventId = subEventId;
171 : {
172 6 : const std::unique_lock<std::mutex> lk(g_mapMutex);
173 6 : const auto iter = g_asyncTaskMap.find(info);
174 6 : if (iter != g_asyncTaskMap.end()) {
175 1 : AICPUE_LOGE("AsyncEventManager RegEventCb failed.");
176 1 : return false;
177 : }
178 5 : g_asyncTaskMap[info] = taskInfo;
179 6 : }
180 :
181 5 : AICPUE_LOGI("AsyncEventManager RegEventCb success, event_id[%u], subeventId[%u], taskId[%lu],"
182 : " streamId[%u], waitType[%u], waitId[%u], opName[%s], startTick[%lu].",
183 : eventId, subEventId, taskInfo.taskId, taskInfo.streamId, static_cast<uint32_t>(taskInfo.waitType),
184 : taskInfo.waitId, taskInfo.opName.c_str(), taskInfo.startTick);
185 5 : return true;
186 8 : }
187 :
188 2 : void AsyncEventManager::UnregEventCb(const uint32_t eventId, const uint32_t subEventId)
189 : {
190 2 : const std::unique_lock<std::mutex> lk(g_mapMutex);
191 : AsyncEventInfo info;
192 2 : info.eventId = eventId;
193 2 : info.subEventId = subEventId;
194 2 : const auto iter = g_asyncTaskMap.find(info);
195 2 : if (iter == g_asyncTaskMap.end()) {
196 2 : AICPUE_LOGW("AsyncEventManager pass call UnregEventCb with eventId[%u], subEventId[%u]", eventId, subEventId);
197 6 : for (auto& kv : g_asyncTaskMap) {
198 4 : AICPUE_LOGI("UnregEventCb show: eventId[%u], subEventId[%u]", kv.first.eventId, kv.first.eventId);
199 : }
200 2 : return;
201 : }
202 0 : (void)g_asyncTaskMap.erase(iter);
203 0 : AICPUE_LOGI("AsyncEventManager UnregEventCb success, eventId[%u], subEventId[%u]", eventId, subEventId);
204 2 : }
205 :
206 13 : void AsyncEventManager::ProcessEvent(const uint32_t eventId, const uint32_t subEventId, void * const param)
207 : {
208 13 : AICPUE_LOGI("AsyncEventManager proc eventId = %d, subEventId = %d", eventId, subEventId);
209 : AsyncEventInfo info;
210 13 : info.eventId = eventId;
211 13 : info.subEventId = subEventId;
212 13 : EventProcessCallBack taskCb = nullptr;
213 : {
214 13 : const std::unique_lock<std::mutex> lk(g_mapMutex);
215 13 : const auto iter = g_asyncTaskMap.find(info);
216 13 : if (iter == g_asyncTaskMap.end()) {
217 10 : AICPUE_LOGW("AsyncEventManager no async task to deal with.");
218 10 : return;
219 : }
220 3 : taskCb = iter->second.taskCb;
221 3 : iter->second.currentTimes++;
222 3 : if ((iter->second.currentTimes >= iter->second.maxTimes) && (iter->second.maxTimes >= 0)) {
223 3 : (void)g_asyncTaskMap.erase(iter);
224 : }
225 13 : }
226 3 : if (taskCb != nullptr) {
227 3 : taskCb(param);
228 : }
229 3 : AICPUE_LOGI("AsyncEventManager proc end!");
230 3 : return;
231 13 : }
232 :
233 7 : bool AsyncEventManager::RegOpEventCb(const uint32_t eventId, const uint32_t subEventId, const EventProcessCallBack &cb)
234 : const {
235 7 : if (cb == nullptr) {
236 1 : AICPUE_LOGE("AsyncEventManager RegOpEventCb on eventId[%u], subeventId[%u] failed, cb is nullptr.",
237 : eventId, subEventId);
238 1 : return false;
239 : }
240 6 : AsyncEventInfo info = {};
241 6 : info.eventId = eventId;
242 6 : info.subEventId = subEventId;
243 6 : OpInfo opInfo = {};
244 6 : (void)aicpu::GetTaskAndStreamId(opInfo.taskId, opInfo.streamId);
245 6 : opInfo.threadIndex = aicpu::GetAicpuThreadIndex();
246 : {
247 6 : const std::unique_lock<std::mutex> lk(g_opMapMutex);
248 6 : auto &opMap = g_opAsyncTaskMap[info];
249 6 : const auto iter = opMap.find(opInfo);
250 6 : if (iter != opMap.end()) {
251 1 : AICPUE_LOGE("AsyncEventManager RegOpEventCb failed for streamId[%u], taskId[%lu], threadIndex[%u]"
252 : " has been registered on eventId[%u], subeventId[%u].",
253 : opInfo.streamId, opInfo.taskId, opInfo.threadIndex, eventId, subEventId);
254 1 : return false;
255 : }
256 5 : opMap[opInfo] = cb;
257 6 : }
258 :
259 5 : AICPUE_LOGI("AsyncEventManager RegOpEventCb success, event_id[%u], subeventId[%u], taskId[%lu], streamId[%u],"
260 : " threadIndex[%u].", eventId, subEventId, opInfo.taskId, opInfo.streamId, opInfo.threadIndex);
261 5 : return true;
262 : }
263 :
264 7 : void AsyncEventManager::UnregOpEventCb(const uint32_t eventId, const uint32_t subEventId) const
265 : {
266 7 : const std::unique_lock<std::mutex> lk(g_opMapMutex);
267 7 : AsyncEventInfo info = {};
268 7 : info.eventId = eventId;
269 7 : info.subEventId = subEventId;
270 7 : auto iter = g_opAsyncTaskMap.find(info);
271 7 : if (iter == g_opAsyncTaskMap.end()) {
272 1 : AICPUE_LOGW("AsyncEventManager pass call UnregOpEventCb with eventId[%u], subEventId[%u]", eventId, subEventId);
273 1 : return;
274 : }
275 :
276 6 : OpInfo opInfo = {};
277 6 : (void)aicpu::GetTaskAndStreamId(opInfo.taskId, opInfo.streamId);
278 6 : opInfo.threadIndex = aicpu::GetAicpuThreadIndex();
279 6 : auto &opMap = iter->second;
280 6 : const auto opIter = opMap.find(opInfo);
281 6 : if (opIter == opMap.end()) {
282 1 : AICPUE_LOGW("AsyncEventManager pass call UnregOpEventCb with streamId[%u], taskId[%lu], threadIndex[%u].",
283 : opInfo.streamId, opInfo.taskId, opInfo.threadIndex);
284 1 : return;
285 : }
286 5 : (void)opMap.erase(opIter);
287 5 : if (opMap.empty()) {
288 2 : (void)g_opAsyncTaskMap.erase(iter);
289 : }
290 5 : AICPUE_LOGI("AsyncEventManager UnregEventCb success, eventId[%u], subEventId[%u], streamId[%u], taskId[%lu],"
291 : " threadIndex[%u].", eventId, subEventId, opInfo.streamId, opInfo.taskId, opInfo.threadIndex);
292 7 : }
293 :
294 5 : void AsyncEventManager::ProcessOpEvent(const uint32_t eventId, const uint32_t subEventId, void * const param) const
295 : {
296 5 : AICPUE_LOGI("AsyncEventManager ProcessOpEvent eventId = %u, subEventId = %u", eventId, subEventId);
297 5 : AsyncEventInfo info = {};
298 5 : info.eventId = eventId;
299 5 : info.subEventId = subEventId;
300 : {
301 5 : const std::unique_lock<std::mutex> lk(g_opMapMutex);
302 5 : const auto iter = g_opAsyncTaskMap.find(info);
303 5 : if (iter == g_opAsyncTaskMap.end()) {
304 4 : AICPUE_LOGW("AsyncEventManager no async task to deal with.");
305 4 : return;
306 : }
307 1 : const auto &opMap = iter->second;
308 3 : for (auto &kv : opMap) {
309 2 : const auto &opInfo = kv.first;
310 2 : EventProcessCallBack taskCb = kv.second;
311 2 : AICPUE_LOGI("AsyncEventManager ProcessOpEvent for streamId[%u], taskId[%lu], threadIndex[%u]",
312 : opInfo.streamId, opInfo.taskId, opInfo.threadIndex);
313 2 : taskCb(param);
314 2 : }
315 5 : }
316 1 : AICPUE_LOGI("AsyncEventManager ProcessOpEvent eventId = %u, subEventId = %u end!", eventId, subEventId);
317 1 : return;
318 : }
319 : } // namespace aicpu
320 :
321 :
322 2 : void AicpuNotifyWait(void *notifyParam, const uint32_t paramLen)
323 : {
324 2 : aicpu::AsyncEventManager::GetInstance().NotifyWait(notifyParam, paramLen);
325 2 : return;
326 : }
327 :
328 1 : bool AicpuRegEventCb(const uint32_t eventId, const uint32_t subEventId, const aicpu::EventProcessCallBack &cb)
329 : {
330 1 : return aicpu::AsyncEventManager::GetInstance().RegEventCb(eventId, subEventId, cb);
331 : }
332 :
333 0 : bool AicpuRegEventCbWithTimes(const uint32_t eventId, const uint32_t subEventId, const aicpu::EventProcessCallBack &cb,
334 : const int32_t times)
335 : {
336 0 : return aicpu::AsyncEventManager::GetInstance().RegEventCb(eventId, subEventId, cb, times);
337 : }
338 :
339 1 : void AicpuUnregEventCb(const uint32_t eventId, const uint32_t subEventId)
340 : {
341 1 : aicpu::AsyncEventManager::GetInstance().UnregEventCb(eventId, subEventId);
342 1 : }
343 :
344 3 : bool AicpuRegOpEventCb(const uint32_t eventId, const uint32_t subEventId, const aicpu::EventProcessCallBack &cb)
345 : {
346 3 : return aicpu::AsyncEventManager::GetInstance().RegOpEventCb(eventId, subEventId, cb);
347 : }
348 :
349 3 : void AicpuUnregOpEventCb(const uint32_t eventId, const uint32_t subEventId)
350 : {
351 3 : aicpu::AsyncEventManager::GetInstance().UnregOpEventCb(eventId, subEventId);
352 3 : }
|
