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 : #ifndef AICPUSD_MODEL_H
12 : #define AICPUSD_MODEL_H
13 :
14 : #include <atomic>
15 : #include <cstdint>
16 : #include <unordered_map>
17 : #include <vector>
18 : #include <unordered_set>
19 : #include <thread>
20 : #include <list>
21 : #include <mutex>
22 : #include "aicpusd_status.h"
23 : #include "aicpusd_common.h"
24 : #include "aicpusd_info.h"
25 : #include "aicpu_task_struct.h"
26 : #include "aicpusd_util.h"
27 : #include "aicpu_event_struct.h"
28 :
29 : #define AICPUSD_EXCEPTION_CATCH(expr0, expr1) \
30 : try { \
31 : (expr0); \
32 : } catch (std::exception & e) { \
33 : aicpusd_err("Error reason:%s", e.what()); \
34 : expr1; \
35 : }
36 :
37 : namespace AicpuSchedule {
38 :
39 : constexpr int64_t INVALID_TABLE_ID = -1;
40 : class AicpuStream {
41 : public:
42 24 : AicpuStream() = default;
43 :
44 24 : ~AicpuStream() = default;
45 :
46 : void InitAicpuStream(const uint32_t streamId, const std::vector<const AicpuTaskInfo*>& tasks);
47 :
48 : int32_t ExecuteNextTask(const RunContext& runContext, bool& streamEnd);
49 :
50 : void ResetToStart();
51 :
52 : void ResetTasks();
53 :
54 : void ShowProgress();
55 :
56 : int32_t AttachReportStatusQueue();
57 :
58 : private:
59 : static int32_t ExecuteTask(const AicpuTaskInfo& kernelTaskInfo, const RunContext& taskContext);
60 : static int32_t ConvertToTsKernel(const AicpuTaskInfo& kernelTaskInfo, aicpu::HwtsTsKernel& aicpufwKernelInfo);
61 :
62 : std::mutex mutexForStream_;
63 : uint32_t streamId_ = 0U;
64 : size_t nextTaskIndex_ = 0LU;
65 : std::vector<AicpuTaskInfo> tasks_;
66 : };
67 :
68 : class QueueMbufStore {
69 : public:
70 15 : QueueMbufStore() = default;
71 :
72 : ~QueueMbufStore();
73 :
74 : bool Init(const size_t queueNum);
75 :
76 : bool Store(const size_t qIndex, Mbuf* const mbuf, std::map<size_t, uint64_t>& gCntList);
77 :
78 : bool IsReady() const;
79 :
80 2 : inline uint64_t BirthDay() const { return birthTimeStamp_; }
81 :
82 : bool Consume(Mbuf*** mbufPptr, std::map<size_t, uint64_t>& gCntList);
83 :
84 : void Free(std::map<size_t, uint64_t>* const gCntList = nullptr);
85 :
86 : bool IsEmpty() const;
87 :
88 : private:
89 : uint64_t birthTimeStamp_{0U};
90 : std::vector<std::list<Mbuf*>> queuesLists_;
91 : };
92 :
93 : enum class GatherResult : int32_t { UN_SELECTED = 0, SELECTED, FAKE_SELECTED };
94 :
95 : enum class StoreResult : int32_t { SUCCESS_STORE = 0, FAIL_STORE, ABORT_STORE };
96 :
97 : enum class ExceptionAction : uint32_t { ADD = 0, EXPIRE = 1 };
98 :
99 : struct StepIdInfo {
100 : uint64_t* stepIdAddr;
101 : uint32_t stepId;
102 :
103 : StepIdInfo() : stepIdAddr(nullptr), stepId(0U){};
104 5404 : StepIdInfo(uint64_t* addr, uint32_t id) : stepIdAddr(addr), stepId(id){};
105 :
106 1 : std::string DebugString() const
107 : {
108 1 : std::stringstream ss;
109 : ss << "Step id info. "
110 1 : << "stepId=" << stepId << std::endl;
111 :
112 2 : return ss.str();
113 1 : }
114 : };
115 :
116 : class AicpuModel {
117 : public:
118 5398 : AicpuModel() = default;
119 :
120 5398 : ~AicpuModel() { (void)pthread_rwlock_destroy(&rwlockForStream_); }
121 :
122 : int32_t Exit();
123 :
124 : int32_t ModelLoad(const AicpuModelInfo* const modelInfo, const ModelCfgInfo* const cfgInfo = nullptr);
125 :
126 : int32_t ModelExecute();
127 :
128 : int32_t TaskReport();
129 :
130 : int32_t ModelAbort();
131 :
132 : int32_t ModelDestroy();
133 :
134 : int32_t EndGraph();
135 :
136 : int32_t ActiveStream(const uint32_t streamId);
137 :
138 : int32_t RecoverStream(const uint32_t streamId);
139 :
140 : int32_t ModelRepeat();
141 :
142 : void ProcessModelException(const uint32_t modelId) const;
143 :
144 19 : uint32_t GetModelTsId() const { return modelTsId_; }
145 :
146 19 : void SetModelTransId(const uint64_t transId) { modelTransId_ = transId; }
147 :
148 3 : uint64_t GetModelTransId() const { return modelTransId_; }
149 :
150 : int32_t UnGardModelBuf(Mbuf* const mbuf);
151 :
152 2 : uint32_t GetReportStmId() const { return reportStream_; }
153 :
154 25658 : bool IsValid() const { return isValid; }
155 :
156 8 : AicpuModelStatus GetModelStatus() const { return modelStatus_; }
157 :
158 22 : ModelPrepareData& GetModelPrepareData() { return prepareData_; }
159 :
160 5 : ModelPostpareData& GetModelPostpareData() { return postpareData_; }
161 :
162 22 : std::vector<void*>& GetInputDataPtrs() { return inputDataPtrs_; }
163 :
164 1 : void SetModelEndOfSequence() { endOfSequence_ = true; }
165 :
166 2 : void ReSetModelEndOfSequence() { endOfSequence_ = false; }
167 :
168 11 : bool IsEndOfSequence() const { return endOfSequence_.load(); }
169 :
170 12 : void SetExtModelId(const uint32_t extId) { extModelId_ = extId; }
171 :
172 7 : uint64_t GetIteratorId() const { return iteratorCount_; }
173 :
174 12 : inline StepIdInfo GetStepIdInfo() const { return stepIdInfo_; }
175 :
176 5 : inline void SetStepIdInfo(const StepIdInfo& info)
177 : {
178 5 : stepIdInfo_ = info;
179 5 : return;
180 : }
181 :
182 13 : inline bool GetHeadNodeFlag() { return headNodeFlag_; }
183 :
184 5 : inline void SetHeadNodeFlag(const bool val) { headNodeFlag_ = val; }
185 :
186 14 : std::vector<bool>& MutableInputsIsDequeue() { return inputsIsDequeue_; }
187 :
188 18 : void SetModelRetCode(const int32_t retVal) { retCode_ = retVal; }
189 :
190 92 : int32_t GetModelRetCode() const { return retCode_.load(); }
191 :
192 8 : bool AbnormalNeedBreak() const { return abnormalBreak_.load(); }
193 :
194 1 : bool AbnormalNeedEnqueue() const { return abnormalEnqueue_.load(); }
195 :
196 24 : bool AbnormalEnabled() const { return abnormalEnabled_.load(); }
197 :
198 3 : inline uint32_t GetId() const { return modelId_; }
199 :
200 : void WaitReleaseThreadsFinish();
201 :
202 3 : inline void SetNullDataFlag(const bool val) { nullDataFlag_ = val; }
203 :
204 18 : inline bool GetNullDataFlag() const { return nullDataFlag_.load(); }
205 :
206 : inline uint32_t GetInputQueueCount() const { return static_cast<uint32_t>(inputQueueIds_.size()); }
207 :
208 0 : inline bool HasQueue(uint32_t queueId) const { return inputQueueIds_.count(queueId) > 0U; }
209 :
210 : StoreResult StoreDequedMbuf(
211 : const uint64_t transId, const uint32_t routeLabel, const size_t qIndex, Mbuf* const mbuf,
212 : const uint32_t queueCount);
213 :
214 : GatherResult SelectGatheredMbuf(Mbuf*** const mbufPptr, const int32_t timeOut, const uint32_t cacheNum);
215 :
216 : void RecordLockedTable(const uint32_t tableId);
217 :
218 : void ClearLockedTable(const uint32_t tableId);
219 :
220 : bool IsTableLocked(const uint32_t tableId);
221 :
222 : void ClearAllLockedTable();
223 :
224 7 : inline void SetTableTryLock(const int64_t tableId) { tableTryLock_ = tableId; }
225 :
226 6 : inline int64_t GetTableTryLock() const { return tableTryLock_; }
227 :
228 : uint32_t& GetInputConsumeNumRef();
229 :
230 3 : inline uint32_t GetActiveStreamNum() const { return activeStreamNum_; }
231 :
232 7 : inline void IncreaseActiveStreamNum() { activeStreamNum_++; }
233 :
234 5 : bool GetModelDestroyStatus() const { return isDestroyModel_; }
235 :
236 : int32_t ModelStop();
237 :
238 : int32_t ModelRestart();
239 :
240 : int32_t ModelClearInput();
241 :
242 : size_t GetCurDequeIndex(const size_t qCnt);
243 :
244 0 : inline void ResetStaticNNModelOutputIndex() { staticNNCurOutIndex_ = 0U; }
245 :
246 0 : inline void IncreaseStaticNNModelOutputIndex() { staticNNCurOutIndex_++; }
247 :
248 0 : inline uint32_t GetCurStaticNNModelOutputIndex() const { return staticNNCurOutIndex_; }
249 :
250 : int32_t ProcessDataException(const uint64_t transId, const uint32_t type);
251 :
252 : bool IsTransIdException(const uint64_t transId);
253 :
254 : void GetExcptionTransIdsToClear(std::vector<int64_t>& excptionTransIdsToClear);
255 :
256 : void UpdateExcptionTransIdsStatus(const std::vector<int64_t> excptionTransIdsCleared);
257 :
258 : void ClearExceptionStore();
259 :
260 : private:
261 : // it is used to store data of the model parepare process so that the process can continue after interruption.
262 : ModelPrepareData prepareData_;
263 : // it is used to store data of the model postpare process so that the process can continue after interruption.
264 : ModelPostpareData postpareData_;
265 :
266 : int32_t ExecuteStream(const uint32_t streamId, const bool executeInline);
267 :
268 : void ActiveOtherAicpuStreams();
269 :
270 : /**
271 : * @brief GetStream by streamId
272 : * attention: must get rwlockForStream_ lock out side.
273 : * @param streamId stream id
274 : * @return aicpu stream
275 : */
276 : AicpuStream* GetStreamByStreamId(const uint32_t streamId);
277 :
278 : /**
279 : * @brief check if allow operate, if allow update status to operate dst status.
280 : * @param operate model operate
281 : * @return AICPU_SCHEDULE_OK:success, other failed.
282 : */
283 : int32_t CheckOperateAndUpdateStatus(const AicpuModelOperate operate);
284 :
285 : /**
286 : * @brief check if allow operate.
287 : * @param operate model operate
288 : * @return AICPU_SCHEDULE_OK:success, other failed.
289 : */
290 : int32_t CheckOperate(const AicpuModelOperate operate);
291 :
292 : /**
293 : * @brief Reset model for execute.
294 : * @return AICPU_SCHEDULE_OK:success, other failed.
295 : */
296 : int32_t ResetModelForExecute();
297 :
298 : /**
299 : * @brief release model tmp resource.
300 : * @return AICPU_SCHEDULE_OK:success, other failed.
301 : */
302 : int32_t ReleaseModelResource();
303 :
304 : int32_t LoadStreamAndTask(const AicpuModelInfo* const modelInfo);
305 :
306 : int32_t LoadQueueInfo(const AicpuModelInfo* const modelInfo);
307 :
308 : /**
309 : * @brief clear all load info
310 : */
311 : void ClearLoadInfo();
312 :
313 : __attribute__((visibility("hidden"))) void LoadWaitNotifyId(
314 : const AicpuTaskInfo& aicpuTaskInfo, std::unordered_set<size_t>& waitNotifyIdSet) const;
315 :
316 : GatherResult GatherDequedMbuf(
317 : Mbuf*** mbufPptr, std::pair<uint64_t, uint32_t>& mbufKey, const int32_t timeOutMs, const uint32_t cacheNum);
318 : void ClearDequedMbuf(const uint64_t transId, const uint32_t routeLabel);
319 : void ClearGatheredMbuf();
320 :
321 : void UpdateModelRetCode(const int32_t retCode);
322 :
323 : int32_t AttachReportStatusQueue();
324 :
325 : int32_t ModelClearInputQueues(const std::unordered_set<size_t>& queueIds, const uint32_t deviceId) const;
326 :
327 : bool IsNewVersion();
328 : void SetVersion(bool isNewVersion);
329 :
330 : private:
331 : static const bool modelOperatePermission[static_cast<int32_t>(AicpuModelStatus::MODEL_STATUS_MAX)]
332 : [static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)];
333 : static const AicpuModelStatus operateNextStatus[static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)];
334 :
335 : volatile bool isValid = false;
336 : // model mutex, guard for operate and status
337 : std::mutex mutexForModel_;
338 : uint32_t modelId_ = INVALID_NUMBER;
339 : uint32_t modelTsId_ = INVALID_NUMBER;
340 : uint64_t modelTransId_ = UINT64_MAX;
341 : uint32_t modelS0Stream_ = INVALID_NUMBER;
342 : uint32_t reportStream_ = INVALID_NUMBER;
343 :
344 : // model status, guard by mutexForModelStatus_.
345 : AicpuModelStatus modelStatus_ = AicpuModelStatus::MODEL_STATUS_UNINIT;
346 : std::mutex mutexForModelStatus_;
347 :
348 : // all streams include aicpu stream and ts stream
349 : std::vector<StreamInfo> allStreams_;
350 : // model streams, guard by rwlockForStream_
351 : std::unordered_map<uint32_t, AicpuStream> aicpuStreams_;
352 : // it is used for stream.
353 : pthread_rwlock_t rwlockForStream_ = PTHREAD_RWLOCK_INITIALIZER;
354 :
355 : // it is used to store model notifyIds
356 : std::unordered_set<size_t> modelNotifyId_;
357 : // the mutex is used for modelNotifyId_.
358 : std::mutex mutexForModelNotifyId_;
359 :
360 : // it is used to store the queue which is subscribed event in one model.
361 : std::vector<QueInfo> queueEventSubscribedInfo_;
362 : // it is used to store the input queue id which is subscribed event in one model.
363 : std::unordered_set<size_t> inputQueueIds_;
364 : // it is used to indicate whether the input is dequeued, size is equal to input queue size
365 : std::vector<bool> inputsIsDequeue_;
366 : // it is used to store the output queue id which is subscribed event in one model.
367 : std::unordered_set<size_t> outputQueueIds_;
368 : // the mutex is used to in storing the relationship of model and queueInfo.
369 : std::mutex mutexForQueueEventSubscribed_;
370 :
371 : // input data pointer
372 : std::vector<void*> inputDataPtrs_;
373 : // the mutex is used to operater mbuff list
374 : std::mutex mutexForMbuffList_;
375 : // it is used to record status(end of sequence) of the model.
376 : std::atomic<bool> endOfSequence_{false};
377 : // it is used to record status of the model.
378 : std::atomic<int32_t> retCode_{0};
379 : // it is used to identify whether to exist.
380 : std::atomic<bool> abnormalBreak_{false};
381 : // it is used to identify whether to enqueue error flag.
382 : std::atomic<bool> abnormalEnqueue_{false};
383 : // The input mbuf may not be initialized. As a result, the retcode of the model is incorrect.
384 : // You can determine whether the retcode of the model is available based on this flag.
385 : std::atomic<bool> abnormalEnabled_{false};
386 : // it is used to record model execute times
387 : uint64_t iteratorCount_ = 0UL;
388 : // record model step id info
389 : StepIdInfo stepIdInfo_{nullptr, 0U};
390 : // head node flag in nn
391 : bool headNodeFlag_ = false;
392 : // ge model id
393 : uint32_t extModelId_ = INVALID_NUMBER;
394 : // for loadModelWithEvent: input bufpool
395 : // for loadModelWithEvent: output bufpool
396 : // for loadModelWithEvent: hccl tag
397 : // for loadModelWithEmbedding: ps id
398 : int32_t psId_{-1};
399 : // for Embedding counter filter feature
400 : bool isSupportCounterFilter_{false};
401 : std::vector<uint32_t> otherAicpuStreams_;
402 : // if dataFlag is 0 and has EOS flag, then only transport data to output. Do not active model.
403 : std::atomic<bool> nullDataFlag_{false};
404 : std::unordered_map<uint64_t, std::unordered_map<uint32_t, QueueMbufStore>> gatheredMbuf_;
405 : std::unordered_map<uint32_t, uint32_t> tableLocked_;
406 : int64_t tableTryLock_{INVALID_TABLE_ID};
407 : // num of times when report status fail
408 : uint32_t inputConsumeNum_{0U};
409 : uint32_t activeStreamNum_{0U};
410 : bool isDestroyModel_ = false;
411 : std::unordered_set<size_t> inputMsgQueueIds_;
412 : std::unordered_set<size_t> outputMsgQueueIds_;
413 : uint32_t staticNNCurOutIndex_{0U};
414 : std::mutex mutexForAsyncTask_;
415 : std::map<size_t, uint64_t> gatheredMbufCntList_;
416 : std::mutex mutexForExceptionTrans_;
417 : std::unordered_map<uint32_t, bool> exceptionTranses_;
418 : };
419 : } // namespace AicpuSchedule
420 : #endif // MAIN_AICPUSD_MODEL_H
|
