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
81 : {
82 2 : return birthTimeStamp_;
83 : }
84 :
85 : bool Consume(Mbuf ***mbufPptr, std::map<size_t, uint64_t> &gCntList);
86 :
87 : void Free(std::map<size_t, uint64_t> *const gCntList = nullptr);
88 :
89 : bool IsEmpty() const;
90 :
91 : private:
92 : uint64_t birthTimeStamp_{0U};
93 : std::vector<std::list<Mbuf *>> queuesLists_;
94 : };
95 :
96 : enum class GatherResult : int32_t {
97 : UN_SELECTED = 0,
98 : SELECTED,
99 : FAKE_SELECTED
100 : };
101 :
102 : enum class StoreResult : int32_t {
103 : SUCCESS_STORE = 0,
104 : FAIL_STORE,
105 : ABORT_STORE
106 : };
107 :
108 : enum class ExceptionAction : uint32_t {
109 : ADD = 0,
110 : EXPIRE = 1
111 : };
112 :
113 : struct StepIdInfo {
114 : uint64_t *stepIdAddr;
115 : uint32_t stepId;
116 :
117 : StepIdInfo() : stepIdAddr(nullptr), stepId(0U) {};
118 5401 : StepIdInfo(uint64_t *addr, uint32_t id) : stepIdAddr(addr), stepId(id) {};
119 :
120 1 : std::string DebugString() const
121 : {
122 1 : std::stringstream ss;
123 : ss << "Step id info. "
124 1 : << "stepId=" << stepId << std::endl;
125 :
126 2 : return ss.str();
127 1 : }
128 : };
129 :
130 : class AicpuModel {
131 : public:
132 5395 : AicpuModel() = default;
133 :
134 5395 : ~AicpuModel()
135 : {
136 5395 : (void)pthread_rwlock_destroy(&rwlockForStream_);
137 5395 : }
138 :
139 : int32_t Exit();
140 :
141 : int32_t ModelLoad(const AicpuModelInfo * const modelInfo, const ModelCfgInfo * const cfgInfo = nullptr);
142 :
143 : int32_t ModelExecute();
144 :
145 : int32_t TaskReport();
146 :
147 : int32_t ModelAbort();
148 :
149 : int32_t ModelDestroy();
150 :
151 : int32_t EndGraph();
152 :
153 : int32_t ActiveStream(const uint32_t streamId);
154 :
155 : int32_t RecoverStream(const uint32_t streamId);
156 :
157 : int32_t ModelRepeat();
158 :
159 : void ProcessModelException(const uint32_t modelId) const;
160 :
161 19 : uint32_t GetModelTsId() const
162 : {
163 19 : return modelTsId_;
164 : }
165 :
166 19 : void SetModelTransId(const uint64_t transId)
167 : {
168 19 : modelTransId_ = transId;
169 19 : }
170 :
171 3 : uint64_t GetModelTransId() const
172 : {
173 3 : return modelTransId_;
174 : }
175 :
176 : int32_t UnGardModelBuf(Mbuf *const mbuf);
177 :
178 2 : uint32_t GetReportStmId() const
179 : {
180 2 : return reportStream_;
181 : }
182 :
183 25658 : bool IsValid() const
184 : {
185 25658 : return isValid;
186 : }
187 :
188 8 : AicpuModelStatus GetModelStatus() const
189 : {
190 8 : return modelStatus_;
191 : }
192 :
193 22 : ModelPrepareData &GetModelPrepareData()
194 : {
195 22 : return prepareData_;
196 : }
197 :
198 5 : ModelPostpareData &GetModelPostpareData()
199 : {
200 5 : return postpareData_;
201 : }
202 :
203 22 : std::vector<void *> &GetInputDataPtrs()
204 : {
205 22 : return inputDataPtrs_;
206 : }
207 :
208 1 : void SetModelEndOfSequence()
209 : {
210 1 : endOfSequence_ = true;
211 1 : }
212 :
213 2 : void ReSetModelEndOfSequence()
214 : {
215 2 : endOfSequence_ = false;
216 2 : }
217 :
218 11 : bool IsEndOfSequence() const
219 : {
220 11 : return endOfSequence_.load();
221 : }
222 :
223 12 : void SetExtModelId(const uint32_t extId)
224 : {
225 12 : extModelId_ = extId;
226 12 : }
227 :
228 7 : uint64_t GetIteratorId() const
229 : {
230 7 : return iteratorCount_;
231 : }
232 :
233 12 : inline StepIdInfo GetStepIdInfo() const
234 : {
235 12 : return stepIdInfo_;
236 : }
237 :
238 5 : inline void SetStepIdInfo(const StepIdInfo &info)
239 : {
240 5 : stepIdInfo_ = info;
241 5 : return;
242 : }
243 :
244 13 : inline bool GetHeadNodeFlag()
245 : {
246 13 : return headNodeFlag_;
247 : }
248 :
249 5 : inline void SetHeadNodeFlag(const bool val)
250 : {
251 5 : headNodeFlag_ = val;
252 5 : }
253 :
254 14 : std::vector<bool> &MutableInputsIsDequeue()
255 : {
256 14 : return inputsIsDequeue_;
257 : }
258 :
259 18 : void SetModelRetCode(const int32_t retVal)
260 : {
261 18 : retCode_ = retVal;
262 18 : }
263 :
264 46 : int32_t GetModelRetCode() const
265 : {
266 92 : return retCode_.load();
267 : }
268 :
269 8 : bool AbnormalNeedBreak() const
270 : {
271 8 : return abnormalBreak_.load();
272 : }
273 :
274 1 : bool AbnormalNeedEnqueue() const
275 : {
276 1 : return abnormalEnqueue_.load();
277 : }
278 :
279 24 : bool AbnormalEnabled() const
280 : {
281 24 : return abnormalEnabled_.load();
282 : }
283 :
284 3 : inline uint32_t GetId() const
285 : {
286 3 : return modelId_;
287 : }
288 :
289 : void WaitReleaseThreadsFinish();
290 :
291 3 : inline void SetNullDataFlag(const bool val)
292 : {
293 3 : nullDataFlag_ = val;
294 3 : }
295 :
296 18 : inline bool GetNullDataFlag() const
297 : {
298 18 : return nullDataFlag_.load();
299 : }
300 :
301 : inline uint32_t GetInputQueueCount() const
302 : {
303 : return static_cast<uint32_t>(inputQueueIds_.size());
304 : }
305 :
306 0 : inline bool HasQueue(uint32_t queueId) const
307 : {
308 0 : return inputQueueIds_.count(queueId) > 0U;
309 : }
310 :
311 : StoreResult StoreDequedMbuf(const uint64_t transId, const uint32_t routeLabel, const size_t qIndex, Mbuf *const mbuf,
312 : const uint32_t queueCount);
313 :
314 : GatherResult SelectGatheredMbuf(Mbuf ***const mbufPptr, const int32_t timeOut, const uint32_t cacheNum);
315 :
316 : void RecordLockedTable(const uint32_t tableId);
317 :
318 : void ClearLockedTable(const uint32_t tableId);
319 :
320 : bool IsTableLocked(const uint32_t tableId);
321 :
322 : void ClearAllLockedTable();
323 :
324 7 : inline void SetTableTryLock(const int64_t tableId)
325 : {
326 7 : tableTryLock_ = tableId;
327 7 : }
328 :
329 6 : inline int64_t GetTableTryLock() const
330 : {
331 6 : return tableTryLock_;
332 : }
333 :
334 : uint32_t &GetInputConsumeNumRef();
335 :
336 3 : inline uint32_t GetActiveStreamNum() const
337 : {
338 3 : return activeStreamNum_;
339 : }
340 :
341 7 : inline void IncreaseActiveStreamNum()
342 : {
343 7 : activeStreamNum_++;
344 7 : }
345 :
346 5 : bool GetModelDestroyStatus() const
347 : {
348 5 : return isDestroyModel_;
349 : }
350 :
351 : int32_t ModelStop();
352 :
353 : int32_t ModelRestart();
354 :
355 : int32_t ModelClearInput();
356 :
357 : size_t GetCurDequeIndex(const size_t qCnt);
358 :
359 0 : inline void ResetStaticNNModelOutputIndex()
360 : {
361 0 : staticNNCurOutIndex_ = 0U;
362 0 : }
363 :
364 0 : inline void IncreaseStaticNNModelOutputIndex()
365 : {
366 0 : staticNNCurOutIndex_++;
367 0 : }
368 :
369 0 : inline uint32_t GetCurStaticNNModelOutputIndex() const
370 : {
371 0 : return staticNNCurOutIndex_;
372 : }
373 :
374 : int32_t ProcessDataException(const uint64_t transId, const uint32_t type);
375 :
376 : bool IsTransIdException(const uint64_t transId);
377 :
378 : void GetExcptionTransIdsToClear(std::vector<int64_t> &excptionTransIdsToClear);
379 :
380 : void UpdateExcptionTransIdsStatus(const std::vector<int64_t> excptionTransIdsCleared);
381 :
382 : void ClearExceptionStore();
383 : private:
384 : // it is used to store data of the model parepare process so that the process can continue after interruption.
385 : ModelPrepareData prepareData_;
386 : // it is used to store data of the model postpare process so that the process can continue after interruption.
387 : ModelPostpareData postpareData_;
388 :
389 : int32_t ExecuteStream(const uint32_t streamId, const bool executeInline);
390 :
391 : void ActiveOtherAicpuStreams();
392 :
393 : /**
394 : * @brief GetStream by streamId
395 : * attention: must get rwlockForStream_ lock out side.
396 : * @param streamId stream id
397 : * @return aicpu stream
398 : */
399 : AicpuStream *GetStreamByStreamId(const uint32_t streamId);
400 :
401 : /**
402 : * @brief check if allow operate, if allow update status to operate dst status.
403 : * @param operate model operate
404 : * @return AICPU_SCHEDULE_OK:success, other failed.
405 : */
406 : int32_t CheckOperateAndUpdateStatus(const AicpuModelOperate operate);
407 :
408 : /**
409 : * @brief check if allow operate.
410 : * @param operate model operate
411 : * @return AICPU_SCHEDULE_OK:success, other failed.
412 : */
413 : int32_t CheckOperate(const AicpuModelOperate operate);
414 :
415 : /**
416 : * @brief Reset model for execute.
417 : * @return AICPU_SCHEDULE_OK:success, other failed.
418 : */
419 : int32_t ResetModelForExecute();
420 :
421 : /**
422 : * @brief release model tmp resource.
423 : * @return AICPU_SCHEDULE_OK:success, other failed.
424 : */
425 : int32_t ReleaseModelResource();
426 :
427 : int32_t LoadStreamAndTask(const AicpuModelInfo * const modelInfo);
428 :
429 : int32_t LoadQueueInfo(const AicpuModelInfo * const modelInfo);
430 :
431 : /**
432 : * @brief clear all load info
433 : */
434 : void ClearLoadInfo();
435 :
436 : __attribute__((visibility("hidden")))
437 : void LoadWaitNotifyId(const AicpuTaskInfo &aicpuTaskInfo,
438 : std::unordered_set<size_t> &waitNotifyIdSet) const;
439 :
440 : GatherResult GatherDequedMbuf(Mbuf ***mbufPptr, std::pair<uint64_t, uint32_t> &mbufKey,
441 : const int32_t timeOutMs, const uint32_t cacheNum);
442 : void ClearDequedMbuf(const uint64_t transId, const uint32_t routeLabel);
443 : void ClearGatheredMbuf();
444 :
445 : void UpdateModelRetCode(const int32_t retCode);
446 :
447 : int32_t AttachReportStatusQueue();
448 :
449 : int32_t ModelClearInputQueues(const std::unordered_set<size_t> &queueIds, const uint32_t deviceId) const;
450 :
451 : bool IsNewVersion();
452 : void SetVersion(bool isNewVersion);
453 :
454 : private:
455 : static const bool modelOperatePermission[static_cast<int32_t>(AicpuModelStatus::MODEL_STATUS_MAX)]
456 : [static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)];
457 : static const AicpuModelStatus operateNextStatus[static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)];
458 :
459 : volatile bool isValid = false;
460 : // model mutex, guard for operate and status
461 : std::mutex mutexForModel_;
462 : uint32_t modelId_ = INVALID_NUMBER;
463 : uint32_t modelTsId_ = INVALID_NUMBER;
464 : uint64_t modelTransId_ = UINT64_MAX;
465 : uint32_t modelS0Stream_ = INVALID_NUMBER;
466 : uint32_t reportStream_ = INVALID_NUMBER;
467 :
468 : // model status, guard by mutexForModelStatus_.
469 : AicpuModelStatus modelStatus_ = AicpuModelStatus::MODEL_STATUS_UNINIT;
470 : std::mutex mutexForModelStatus_;
471 :
472 : // all streams include aicpu stream and ts stream
473 : std::vector<StreamInfo> allStreams_;
474 : // model streams, guard by rwlockForStream_
475 : std::unordered_map<uint32_t, AicpuStream> aicpuStreams_;
476 : // it is used for stream.
477 : pthread_rwlock_t rwlockForStream_ = PTHREAD_RWLOCK_INITIALIZER;
478 :
479 : // it is used to store model notifyIds
480 : std::unordered_set<size_t> modelNotifyId_;
481 : // the mutex is used for modelNotifyId_.
482 : std::mutex mutexForModelNotifyId_;
483 :
484 : // it is used to store the queue which is subscribed event in one model.
485 : std::vector<QueInfo> queueEventSubscribedInfo_;
486 : // it is used to store the input queue id which is subscribed event in one model.
487 : std::unordered_set<size_t> inputQueueIds_;
488 : // it is used to indicate whether the input is dequeued, size is equal to input queue size
489 : std::vector<bool> inputsIsDequeue_;
490 : // it is used to store the output queue id which is subscribed event in one model.
491 : std::unordered_set<size_t> outputQueueIds_;
492 : // the mutex is used to in storing the relationship of model and queueInfo.
493 : std::mutex mutexForQueueEventSubscribed_;
494 :
495 : // input data pointer
496 : std::vector<void *> inputDataPtrs_;
497 : // the mutex is used to operater mbuff list
498 : std::mutex mutexForMbuffList_;
499 : // it is used to record status(end of sequence) of the model.
500 : std::atomic<bool> endOfSequence_{false};
501 : // it is used to record status of the model.
502 : std::atomic<int32_t> retCode_{0};
503 : // it is used to identify whether to exist.
504 : std::atomic<bool> abnormalBreak_{false};
505 : // it is used to identify whether to enqueue error flag.
506 : std::atomic<bool> abnormalEnqueue_{false};
507 : // The input mbuf may not be initialized. As a result, the retcode of the model is incorrect.
508 : // You can determine whether the retcode of the model is available based on this flag.
509 : std::atomic<bool> abnormalEnabled_{false};
510 : // it is used to record model execute times
511 : uint64_t iteratorCount_ = 0UL;
512 : // record model step id info
513 : StepIdInfo stepIdInfo_{nullptr, 0U};
514 : // head node flag in nn
515 : bool headNodeFlag_ = false;
516 : // ge model id
517 : uint32_t extModelId_ = INVALID_NUMBER;
518 : // for loadModelWithEvent: input bufpool
519 : // for loadModelWithEvent: output bufpool
520 : // for loadModelWithEvent: hccl tag
521 : // for loadModelWithEmbedding: ps id
522 : int32_t psId_{-1};
523 : // for Embedding counter filter feature
524 : bool isSupportCounterFilter_{false};
525 : std::vector<uint32_t> otherAicpuStreams_;
526 : // if dataFlag is 0 and has EOS flag, then only transport data to output. Do not active model.
527 : std::atomic<bool> nullDataFlag_{false};
528 : std::unordered_map<uint64_t, std::unordered_map<uint32_t, QueueMbufStore>> gatheredMbuf_;
529 : std::unordered_map<uint32_t, uint32_t> tableLocked_;
530 : int64_t tableTryLock_{INVALID_TABLE_ID};
531 : // num of times when report status fail
532 : uint32_t inputConsumeNum_ {0U};
533 : uint32_t activeStreamNum_{0U};
534 : bool isDestroyModel_ = false;
535 : std::unordered_set<size_t> inputMsgQueueIds_;
536 : std::unordered_set<size_t> outputMsgQueueIds_;
537 : uint32_t staticNNCurOutIndex_{0U};
538 : std::mutex mutexForAsyncTask_;
539 : std::map<size_t, uint64_t> gatheredMbufCntList_;
540 : std::mutex mutexForExceptionTrans_;
541 : std::unordered_map<uint32_t, bool> exceptionTranses_;
542 : };
543 : }
544 : #endif // MAIN_AICPUSD_MODEL_H
|
