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