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 QUEUE_SCHEDULE_BQS_QUEUE_SCHEDULE_H
12 : #define QUEUE_SCHEDULE_BQS_QUEUE_SCHEDULE_H
13 :
14 : #include <vector>
15 : #include <thread>
16 : #include <atomic>
17 : #include <set>
18 : #include <string>
19 : #include <condition_variable>
20 : #include "driver/ascend_hal.h"
21 : #include "common/bqs_status.h"
22 : #include "common/bqs_msg.h"
23 : #include "bind_relation.h"
24 : namespace bqs {
25 : class QueueSchedule {
26 : public:
27 : /**
28 : * QueueSchedule construct.
29 : * @param deviceId device chip id.
30 : * @param enqueGroupId enqueue group id.
31 : * @param f2nfGroupId full to not full group id.
32 : */
33 108 : explicit QueueSchedule(const InitQsParams& params)
34 108 : : deviceId_(params.deviceId),
35 108 : enqueGroupId_(params.enqueGroupId),
36 108 : f2nfGroupId_(params.f2nfGroupId),
37 108 : running_(false),
38 108 : hasAICPU_(true),
39 108 : isZeroSizeAicpuNum_(false),
40 108 : reschedInterval_(params.reschedInterval),
41 108 : runMode_(params.runMode),
42 108 : qsInitGroupName_(params.qsInitGrpName),
43 108 : initQsParams_(params),
44 216 : abnormalInterval_(0U)
45 108 : {}
46 :
47 : ~QueueSchedule();
48 :
49 : QueueSchedule(const QueueSchedule&) = delete;
50 :
51 : QueueSchedule& operator=(const QueueSchedule&) = delete;
52 :
53 : QueueSchedule(QueueSchedule&&) = delete;
54 :
55 : QueueSchedule& operator=(QueueSchedule&&) = delete;
56 :
57 : public:
58 : /**
59 : * Init and start working.
60 : * @return BQS_STATUS_OK:success, other:failed
61 : */
62 : BqsStatus StartQueueSchedule();
63 :
64 : /* *
65 : * stop working.
66 : */
67 : void StopQueueSchedule();
68 :
69 : /**
70 : * destroy source.
71 : */
72 : void Destroy() const;
73 :
74 : /**
75 : * enqueue event working thread function
76 : * @param threadIndex: thread index
77 : * @param bindCpuIndex: bind cpu index
78 : * @param groupId: wait event group
79 : */
80 : void EnqueueThreadTask(
81 : const uint32_t deviceId, const uint32_t threadIndex, const uint32_t bindCpuIndex, const uint32_t groupId,
82 : const uint32_t index);
83 :
84 : /**
85 : * full to not full working thread function
86 : * @param threadIndex: thread index
87 : * @param bindCpuIndex: bind cpu index
88 : * @param groupId: wait event group
89 : */
90 : void F2NFThreadTask(const uint32_t threadIndex, const uint32_t bindCpuIndex, const uint32_t groupId);
91 :
92 : /**
93 : * daemon thread run function
94 : */
95 : void DaemonThreadTask(const uint32_t index);
96 :
97 : /**
98 : * wait for stop.
99 : */
100 : void WaitForStop();
101 :
102 : void ReportAbnormal() const;
103 :
104 : private:
105 : /**
106 : * handle enqueue event.
107 : * @param threadIndex thread index
108 : * @param event event
109 : */
110 : void ProcessEnqueueEvent(
111 : const uint32_t threadIndex, const event_info& event, const uint32_t index = 0U, const bool procF2NF = false);
112 :
113 : /**
114 : * process enqueue event loop.
115 : * @param threadIndex thread index
116 : * @param groupId groupId
117 : */
118 : void LoopProcessEnqueueEvent(
119 : const uint32_t threadIndex, const uint32_t deviceId, const uint32_t groupId, const uint32_t index);
120 :
121 : /**
122 : * start enqueue thread.
123 : * @param threadNum thread total num
124 : * @param aicpuBeginIndex begin index
125 : */
126 : BqsStatus StartThreadGroup(
127 : const uint32_t threadNum, const uint32_t deviceId, const uint32_t enqueGroupId, const uint32_t index);
128 :
129 : /**
130 : * start enqueue thread.
131 : * @param threadIndex thread id
132 : * @param bindCpuIndex cpu index
133 : */
134 : void BindAicpu(const uint32_t threadIndex, const uint32_t bindCpuIndex);
135 :
136 : /**
137 : * handle enqueue event in daemon thread
138 : */
139 : void DaemonEnqueueEvent(const uint32_t index);
140 :
141 : /**
142 : * schedule data buff for all queue.
143 : * @param dataEnqueue true means data enqueue, false means relation or f2nf enqueue
144 : */
145 : void ScheduleDataBuffAll(const bool dataEnqueue, const uint32_t index = 0U) const;
146 :
147 : /**
148 : * process full entity
149 : * @param entity entity
150 : */
151 : dgw::FsmStatus ProcessDstEntity(const EntityInfo& entity, const uint32_t index) const;
152 :
153 : /**
154 : * init drv event scheduler.
155 : * @return BQS_STATUS_OK: success, other: error
156 : */
157 : BqsStatus InitDrvSchedModule(
158 : const uint32_t deviceId, const uint32_t enqueGroupId, const uint32_t f2nfGroupId) const;
159 :
160 : /**
161 : * process event
162 : * @param event esched event info
163 : */
164 : BqsStatus ProcessEvent(const uint32_t threadIndex, event_info& event, const uint32_t index);
165 :
166 : void CheckIfRecover(
167 : uint32_t& errCount, const char_t* const identity, const uint32_t threadIndex, const uint32_t groupId) const;
168 :
169 : void DynamicSchedule(const uint32_t index) const;
170 :
171 : void ProcessFullToNotFullEvent(const uint32_t index);
172 :
173 : BqsStatus InitExtraSchedule(const std::set<uint32_t>& resDevids, uint32_t threadNum);
174 :
175 : /**
176 : * device chip id.
177 : */
178 : uint32_t deviceId_;
179 :
180 : /**
181 : * enqueue event group id.
182 : */
183 : uint32_t enqueGroupId_;
184 :
185 : /**
186 : * full to not full event group id.
187 : */
188 : uint32_t f2nfGroupId_;
189 :
190 : /**
191 : * thread run flag.
192 : */
193 : volatile bool running_;
194 :
195 : /**
196 : * daemon thread sleep for a while.
197 : */
198 : std::condition_variable daemonWait_;
199 :
200 : /**
201 : * run on aicpu flag.
202 : */
203 : bool hasAICPU_;
204 :
205 : /**
206 : * run on aicpu flag.
207 : */
208 : bool isZeroSizeAicpuNum_;
209 :
210 : /**
211 : * working threads.
212 : */
213 : std::vector<std::thread> workThreads_;
214 :
215 : /**
216 : * enqueue event working mutual exclusion
217 : */
218 : std::atomic_flag queueEventAtomicFlag_ = ATOMIC_FLAG_INIT;
219 :
220 : std::atomic_flag queueEventAtomicFlagExtra_ = ATOMIC_FLAG_INIT;
221 :
222 : /**
223 : * daemon threads wait mtx.
224 : */
225 : std::mutex daemonWaitMtx_;
226 :
227 : /**
228 : * daemon threads reschedule interval
229 : */
230 : uint32_t reschedInterval_;
231 :
232 : /**
233 : * qs run mode
234 : */
235 : bqs::QueueSchedulerRunMode runMode_;
236 :
237 : /**
238 : * group name send by tsd used for qs to attach when start
239 : */
240 : std::string qsInitGroupName_;
241 :
242 : /**
243 : * queue schedule initialization parameters
244 : */
245 : InitQsParams initQsParams_;
246 :
247 : /**
248 : * f2nf event working mutual exclusion
249 : */
250 : std::atomic_flag f2nfEventAtomicFlag_ = ATOMIC_FLAG_INIT;
251 :
252 : bool aicpuFeatureDisableRecvRequestEvent_{false};
253 :
254 : bool aicpuFeatureSetPidPriority_{false};
255 :
256 : uint32_t abnormalInterval_;
257 : };
258 : } // namespace bqs
259 : #endif // QUEUE_SCHEDULE_BQS_QUEUE_SCHEDULE_H
|
