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 "queue_schedule.h"
12 : #include <unistd.h>
13 : #include <sched.h>
14 : #include <csignal>
15 : #include <cerrno>
16 : #include <string>
17 : #include "driver/ascend_hal.h"
18 : #include "server/bqs_server.h"
19 : #include "server/router_server.h"
20 : #include "bind_relation.h"
21 : #include "bind_cpu_utils.h"
22 : #include "statistic_manager.h"
23 : #include "subscribe_manager.h"
24 : #include "queue_manager.h"
25 : #include "profile_manager.h"
26 : #include "entity_manager.h"
27 : #include "hccl_process.h"
28 : #include "common/bqs_util.h"
29 : #include "queue_schedule_sub_module_interface.h"
30 : #include "qs_interface_process.h"
31 : #include "queue_schedule_hal_interface_ref.h"
32 : #include "tsd.h"
33 : #include "schedule_config.h"
34 : #include "dynamic_sched_mgr.hpp"
35 : #include "fsm/state_define.h"
36 : #include "common/bqs_feature_ctrl.h"
37 : #include "qs_args_parser.h"
38 : #include "queue_schedule_feature_ctrl.h"
39 :
40 : namespace bqs {
41 : namespace {
42 : constexpr uint32_t HOST_NAME_MAX_LEN = 128U;
43 : constexpr const char_t *AOS_SD = "AOS_SD";
44 : constexpr const size_t FIRST_ARRAY_INDEX = 0LU;
45 : constexpr const uint32_t DAEMON_WAIT_TIMEOUT = 30U;
46 : constexpr const uint32_t HOST_ENQUEUE_THREAD_NUM = 10U;
47 : constexpr const uint32_t HOST_F2NF_THREAD_NUM = 10U;
48 : // max qos num for hccl event: must greater than thread num in F2NF group
49 : constexpr const uint32_t MAX_QOS_NUM_FOR_HCCL_EVENT = 12U;
50 : // max qos num for f2nf event
51 : constexpr const uint32_t MAX_QOS_NUM_FOR_F2NF_EVENT = 1U;
52 : constexpr const char_t *ENQUEUE_THREAD_NAME_PREFIX = "enqueue_";
53 : constexpr const char_t *F2NF_THREAD_NAME_PREFIX = "f2nf_";
54 : constexpr const char_t *DAEMON_THREAD_NAME_PREFIX = "daemon";
55 : constexpr const uint32_t ERROR_LOG_SAMPLE_INTERVAL = 1000U;
56 :
57 1 : void DynamicScheduleByResponse(const uint32_t key, const uint32_t index,
58 : const std::vector<dgw::DynamicSchedMgr::ResponseInfo> &responses)
59 : {
60 1 : BQS_LOG_INFO("responses size is %zu", responses.size());
61 2 : for (const auto &response : responses) {
62 : dgw::EntityPtr dynamicSrcEntity =
63 1 : dgw::EntityManager::Instance(index).GetSrcEntityByGlobalId(key, response.src.queueLogicId);
64 1 : if (dynamicSrcEntity == nullptr) {
65 0 : BQS_LOG_ERROR("Can't get entity by key[%u], globalId[%u]", key, response.src.queueLogicId);
66 0 : continue;
67 : }
68 :
69 1 : uint32_t updateCount = 0U;
70 1 : const auto &dataResults = response.groupResults;
71 2 : for (const auto groupResult : dataResults) {
72 : dgw::EntityPtr dynamicDstGrpEnity =
73 1 : dgw::EntityManager::Instance(index).GetDstEntityByGlobalId(key, groupResult.logicGroupId);
74 1 : if (dynamicDstGrpEnity == nullptr) {
75 0 : BQS_LOG_ERROR("Can't get entity by key[%u], globalId[%u]", key, groupResult.logicGroupId);
76 0 : continue;
77 : }
78 : const std::vector<dgw::EntityPtr> &entitiesInGroup =
79 1 : dgw::EntityManager::Instance(index).GetEntitiesInGroup(dynamicDstGrpEnity->GetId());
80 1 : if (entitiesInGroup.size() <= groupResult.index) {
81 0 : BQS_LOG_ERROR("Dynamic response's index[%u] is larger than group size[%zu]",
82 : groupResult.index, entitiesInGroup.size());
83 0 : continue;
84 : }
85 1 : const dgw::EntityPtr dynamicDstInGroup = entitiesInGroup[groupResult.index];
86 1 : if (dynamicSrcEntity->UpdateSendObject(dynamicDstGrpEnity, dynamicDstInGroup)) {
87 1 : ++updateCount;
88 : }
89 1 : }
90 1 : BQS_LOG_INFO("updateCount is %u", updateCount);
91 1 : if (updateCount > 0U) {
92 1 : BQS_LOG_INFO("%s processMessage", dynamicSrcEntity->ToString().c_str());
93 1 : dgw::DynamicSchedMgr::GetInstance(index).DynamicSchedDurationEnd(dynamicSrcEntity->GetDynamicReqTime());
94 1 : dgw::InnerMessage msg;
95 1 : msg.msgType = dgw::InnerMsgType::INNER_MSG_PUSH;
96 1 : (void) dynamicSrcEntity->ProcessMessage(msg);
97 : }
98 1 : }
99 1 : }
100 : thread_local static int32_t thread_groupId;
101 : thread_local static int32_t thread_deviceId;
102 :
103 0 : int32_t getGrpId(int32_t tag, int32_t *grpId, int32_t *deviceId)
104 : {
105 : (void) tag;
106 0 : if ((grpId == nullptr) || (deviceId == nullptr)) {
107 0 : return -1;
108 : }
109 0 : *grpId = thread_groupId;
110 0 : *deviceId = thread_deviceId;
111 0 : return 0;
112 : }
113 :
114 : } // namespace
115 :
116 18 : BqsStatus QueueSchedule::StartQueueSchedule()
117 : {
118 18 : auto ret = BqsServer::GetInstance().InitBqsServer(qsInitGroupName_, deviceId_);
119 18 : if (ret != BQS_STATUS_OK) {
120 0 : BQS_LOG_ERROR("BqsServer Init failed, ret=%d.", ret);
121 0 : return ret;
122 : }
123 18 : ret = RouterServer::GetInstance().InitRouterServer(initQsParams_);
124 18 : if (ret != BQS_STATUS_OK) {
125 2 : BQS_LOG_ERROR("RouterServer Init failed, ret=%d.", ret);
126 2 : return ret;
127 : }
128 :
129 16 : GlobalCfg::GetInstance().SetNumaFlag(initQsParams_.numaFlag);
130 16 : GlobalCfg::GetInstance().RecordDeviceId(deviceId_, 0U, enqueGroupId_);
131 16 : reschedInterval_ = (reschedInterval_ == 0U) ? DAEMON_WAIT_TIMEOUT : reschedInterval_;
132 32 : abnormalInterval_ = ((initQsParams_.abnormalInterVal > ABNORMAL_INTERVAL_MIN) &&
133 16 : (initQsParams_.abnormalInterVal < ABNORMAL_INTERVAL_MAX)) ?
134 : initQsParams_.abnormalInterVal : ABNORMAL_INTERVAL_DEFAULT;
135 16 : StatisticManager::GetInstance().StartStatisticManager(
136 16 : abnormalInterval_, initQsParams_.pid, initQsParams_.numaFlag, initQsParams_.deviceIdExtra,
137 : initQsParams_.enqueGroupIdExtra);
138 :
139 16 : char_t hostNameStr[HOST_NAME_MAX_LEN] = {};
140 16 : const int32_t getNameRet = gethostname(&hostNameStr[FIRST_ARRAY_INDEX], HOST_NAME_MAX_LEN);
141 16 : if (getNameRet < 0) {
142 0 : BQS_LOG_ERROR("gethostname failed, ret=%d.", getNameRet);
143 0 : return BQS_STATUS_INNER_ERROR;
144 : }
145 16 : const std::string nameStr(hostNameStr);
146 16 : uint32_t threadNum = 0U;
147 16 : if (nameStr == AOS_SD) {
148 3 : hasAICPU_ = false;
149 3 : threadNum = 1U;
150 : }
151 16 : BQS_LOG_INFO("StartQueueSchedule schedPolicy:[%lu].", initQsParams_.schedPolicy);
152 16 : if (hasAICPU_ && (bqs::RunContext::HOST != bqs::GetRunContext())) {
153 3 : uint32_t aicpuNum = QueueScheduleInterface::GetInstance().GetAiCpuNum();
154 3 : BQS_LOG_RUN_INFO("the number of AICPU cores: %d", aicpuNum);
155 3 : if (aicpuNum == 0U) {
156 2 : isZeroSizeAicpuNum_ = true;
157 2 : threadNum = 1U;
158 : } else {
159 1 : threadNum = aicpuNum;
160 : }
161 : }
162 16 : BQS_LOG_RUN_INFO("Has aicpu:%d, numaFlag:%d, deviceId_:%u.",
163 : static_cast<int32_t>(hasAICPU_), initQsParams_.numaFlag, deviceId_);
164 :
165 16 : aicpuFeatureDisableRecvRequestEvent_ = (bqs::GetRunContext() == bqs::RunContext::HOST) ? false : QSFeatureCtrl::ShouldDisableRecvRequestEvent(deviceId_);
166 16 : aicpuFeatureSetPidPriority_ = (bqs::GetRunContext() == bqs::RunContext::HOST) ? false : QSFeatureCtrl::ShouldSetPidPriority(deviceId_);
167 16 : ret = InitDrvSchedModule(deviceId_, enqueGroupId_, f2nfGroupId_);
168 16 : if (ret != BQS_STATUS_OK) {
169 2 : BQS_LOG_ERROR("InitDrvSchedModule failed, ret=%d.", ret);
170 2 : return ret;
171 : }
172 :
173 14 : ret = QueueManager::GetInstance().InitQueueManager(deviceId_, enqueGroupId_, hasAICPU_, qsInitGroupName_);
174 14 : if (ret != BQS_STATUS_OK) {
175 1 : BQS_LOG_ERROR("QueueManager Init failed, ret=%d.", ret);
176 1 : return ret;
177 : }
178 :
179 13 : std::set<uint32_t> resDevids(initQsParams_.devIdVec.begin(), initQsParams_.devIdVec.end());
180 13 : resDevids.insert(deviceId_);
181 13 : if (initQsParams_.numaFlag) {
182 5 : resDevids.insert(initQsParams_.deviceIdExtra);
183 : }
184 13 : Subscribers::GetInstance().InitSubscribeManagers(resDevids, deviceId_);
185 :
186 13 : ProfileManager::GetInstance(0U).InitProfileManager(deviceId_);
187 13 : dgw::EntityManager::Instance(0U).SetSubscriptionPausePolicy(
188 13 : (initQsParams_.schedPolicy & static_cast<uint64_t>(SchedPolicy::POLICY_UNSUB_F2NF)) == 0UL);
189 :
190 13 : running_ = true;
191 13 : if (BindCpuUtils::InitSem() != BQS_STATUS_OK) {
192 0 : BQS_LOG_ERROR("InitSem failed");
193 0 : return BQS_STATUS_INNER_ERROR;
194 : }
195 :
196 13 : auto threadRet = StartThreadGroup(threadNum, deviceId_, enqueGroupId_, 0U);
197 13 : if (threadRet != BQS_STATUS_OK) {
198 0 : BQS_LOG_ERROR("StartThreadGroup failed");
199 0 : BindCpuUtils::DestroySem();
200 0 : return threadRet;
201 : }
202 :
203 13 : if (initQsParams_.numaFlag) {
204 5 : const auto extraRet = InitExtraSchedule(resDevids, threadNum);
205 5 : if (extraRet != BQS_STATUS_OK) {
206 3 : BQS_LOG_ERROR("InitExtraSchedule failed, ret is %d.", static_cast<int32_t>(extraRet));
207 3 : BindCpuUtils::DestroySem();
208 3 : return extraRet;
209 : }
210 : }
211 :
212 10 : BindCpuUtils::DestroySem();
213 10 : RouterServer::GetInstance().NotifyInitSuccess();
214 10 : return BQS_STATUS_OK;
215 16 : }
216 :
217 5 : BqsStatus QueueSchedule::InitExtraSchedule(const std::set<uint32_t> &resDevids, uint32_t threadNum)
218 : {
219 5 : BqsStatus ret = InitDrvSchedModule(initQsParams_.deviceIdExtra, initQsParams_.enqueGroupIdExtra,
220 : initQsParams_.f2nfGroupIdExtra);
221 5 : if (ret != BQS_STATUS_OK) {
222 1 : BQS_LOG_ERROR("InitDrvSchedModule failed, ret=%d.", static_cast<int32_t>(ret));
223 1 : return ret;
224 : }
225 4 : GlobalCfg::GetInstance().RecordDeviceId(initQsParams_.deviceIdExtra, 1U, initQsParams_.enqueGroupIdExtra);
226 :
227 4 : QueueManager::GetInstance().InitExtra(initQsParams_.deviceIdExtra, initQsParams_.enqueGroupIdExtra);
228 :
229 4 : Subscribers::GetInstance().InitSubscribeManagers(resDevids, initQsParams_.deviceIdExtra);
230 :
231 4 : ProfileManager::GetInstance(1U).InitProfileManager(initQsParams_.deviceIdExtra);
232 :
233 4 : dgw::EntityManager::Instance(1U).SetSubscriptionPausePolicy(
234 4 : (initQsParams_.schedPolicy & static_cast<uint64_t>(SchedPolicy::POLICY_UNSUB_F2NF)) == 0UL);
235 :
236 4 : if (hasAICPU_ && (bqs::RunContext::HOST != bqs::GetRunContext())) {
237 1 : uint32_t aicpuNum = QueueScheduleInterface::GetInstance().GetExtraAiCpuNum();
238 1 : BQS_LOG_RUN_INFO("the number of Extra AICPU cores: %d", aicpuNum);
239 1 : if (aicpuNum == 0U) {
240 1 : aicpuNum = 1U;
241 : }
242 1 : threadNum = aicpuNum;
243 : }
244 :
245 4 : BqsStatus threadRet = StartThreadGroup(threadNum, initQsParams_.deviceIdExtra, initQsParams_.enqueGroupIdExtra, 1U);
246 4 : if (threadRet != BQS_STATUS_OK) {
247 1 : BQS_LOG_ERROR("StartThreadGroup failed");
248 1 : return threadRet;
249 : }
250 :
251 3 : const auto setCallbackRes = HcclSetGrpIdCallback(getGrpId);
252 3 : if (setCallbackRes != HCCL_SUCCESS) {
253 1 : BQS_LOG_ERROR("SetCallback failed, res is %d", static_cast<int32_t>(setCallbackRes));
254 1 : return BQS_STATUS_HCCL_ERROR;
255 : }
256 2 : return BQS_STATUS_OK;
257 : }
258 :
259 15 : BqsStatus QueueSchedule::StartThreadGroup(const uint32_t threadNum, const uint32_t deviceId,
260 : const uint32_t enqueGroupId, const uint32_t index)
261 : {
262 15 : const sighandler_t oldHandler = signal(SIGCHLD, static_cast<sighandler_t>(SIG_DFL));
263 15 : const uint32_t enqueueThreadNum = (bqs::RunContext::HOST == bqs::GetRunContext()) ? HOST_ENQUEUE_THREAD_NUM
264 15 : : threadNum;
265 15 : uint32_t vDevNum = 0U;
266 15 : if ((FeatureCtrl::IsVfModeCheckedByDeviceId(deviceId)) && (&halGetVdevNum != nullptr)) {
267 0 : int32_t ret = halGetVdevNum(&vDevNum);
268 0 : if (ret != 0) {
269 0 : BQS_LOG_ERROR("halGetVdevNum, failed result[%d]", ret);
270 0 : return BQS_STATUS_DRIVER_ERROR;
271 : }
272 : }
273 15 : BQS_LOG_INFO("Get vdev num=[%u] success.", vDevNum);
274 129 : for (uint32_t thIndex = 0U; thIndex < enqueueThreadNum; ++thIndex) {
275 : // create enqueue event thread
276 114 : uint32_t aicpuIndex = 0U;
277 114 : if (bqs::RunContext::HOST != bqs::GetRunContext()) {
278 4 : if (vDevNum > 0U) {
279 0 : aicpuIndex = (index == 0U) ?
280 0 : QueueScheduleInterface::GetInstance().GetAicpuPhysIndexInVfMode(thIndex, deviceId) :
281 0 : QueueScheduleInterface::GetInstance().GetExtraAicpuPhysIndexInVfMode(thIndex, deviceId);
282 : } else {
283 8 : aicpuIndex = (index == 0U) ?
284 4 : QueueScheduleInterface::GetInstance().GetAicpuPhysIndex(deviceId, thIndex) :
285 0 : QueueScheduleInterface::GetInstance().GetExtraAicpuPhysIndex(deviceId, thIndex);
286 : }
287 : }
288 114 : (void) workThreads_.emplace_back(
289 114 : &QueueSchedule::EnqueueThreadTask, this, deviceId, thIndex, aicpuIndex, enqueGroupId, index);
290 : }
291 15 : if (bqs::RunContext::HOST != bqs::GetRunContext()) {
292 4 : (void)workThreads_.emplace_back(&QueueSchedule::DaemonThreadTask, this, index);
293 : }
294 :
295 15 : uint32_t f2nfThreadNum = (bqs::RunContext::HOST == bqs::GetRunContext()) ? HOST_F2NF_THREAD_NUM : threadNum;
296 15 : if (initQsParams_.numaFlag) {
297 : // these event will be processed by enque thread on condition numa, so we need not create f2nf threads
298 7 : f2nfThreadNum = 0U;
299 : }
300 59 : for (uint32_t thIndex = 0U; thIndex < f2nfThreadNum; ++thIndex) {
301 : // create f2nf event thread
302 44 : uint32_t aicpuIndex = 0U;
303 44 : if (bqs::RunContext::HOST != bqs::GetRunContext()) {
304 4 : if (vDevNum > 0U) {
305 0 : aicpuIndex = QueueScheduleInterface::GetInstance().GetAicpuPhysIndexInVfMode(thIndex, deviceId);
306 : } else {
307 4 : aicpuIndex = QueueScheduleInterface::GetInstance().GetAicpuPhysIndex(deviceId_, thIndex);
308 : }
309 : }
310 44 : (void) workThreads_.emplace_back(
311 44 : &QueueSchedule::F2NFThreadTask, this, thIndex, aicpuIndex, f2nfGroupId_);
312 : }
313 :
314 173 : for (uint32_t thIndex = 0U; thIndex < enqueueThreadNum + f2nfThreadNum; ++thIndex) {
315 : // EnqueueThreadTask thread
316 158 : if (BindCpuUtils::WaitSem() != BQS_STATUS_OK) {
317 0 : BQS_LOG_ERROR("WaitSem failed");
318 0 : return BQS_STATUS_INNER_ERROR;
319 : }
320 : }
321 15 : (void)signal(SIGCHLD, oldHandler);
322 15 : return BQS_STATUS_OK;
323 : }
324 :
325 308 : void QueueSchedule::StopQueueSchedule()
326 : {
327 308 : running_ = false;
328 308 : const std::unique_lock<std::mutex> daemonWaitLock(daemonWaitMtx_);
329 308 : daemonWait_.notify_all();
330 308 : StatisticManager::GetInstance().DumpOutProcMemStatInfo();
331 308 : StatisticManager::GetInstance().StopStatisticManager();
332 308 : ProfileManager::GetInstance(0U).Uninit();
333 308 : if (initQsParams_.numaFlag) {
334 10 : ProfileManager::GetInstance(1U).Uninit();
335 : }
336 308 : }
337 :
338 55 : void QueueSchedule::Destroy() const
339 : {
340 55 : QueueManager::GetInstance().Destroy();
341 55 : if (!SubModuleInterface::GetInstance().GetStartFlag()) {
342 53 : (void)halEschedDettachDevice(deviceId_);
343 53 : if (initQsParams_.numaFlag) {
344 5 : (void)halEschedDettachDevice(initQsParams_.deviceIdExtra);
345 : }
346 : } else {
347 2 : BQS_LOG_RUN_INFO("sub module no need process detach main module do it");
348 : }
349 55 : }
350 :
351 114 : void QueueSchedule::EnqueueThreadTask(const uint32_t deviceId, const uint32_t threadIndex, const uint32_t bindCpuIndex,
352 : const uint32_t groupId, const uint32_t index)
353 : {
354 114 : BQS_LOG_INFO("QueueSchedule enqueue thread[%u] start.", threadIndex);
355 114 : BindAicpu(threadIndex, bindCpuIndex);
356 114 : const auto threadName = std::string(ENQUEUE_THREAD_NAME_PREFIX).append(std::to_string(threadIndex));
357 114 : (void)pthread_setname_np(pthread_self(), threadName.c_str());
358 114 : thread_groupId = groupId;
359 114 : thread_deviceId = deviceId;
360 :
361 114 : const uint64_t eventBitmap =
362 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(EVENT_QUEUE_ENQUEUE)) |
363 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(EVENT_QUEUE_FULL_TO_NOT_FULL)) |
364 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_RECV_REQUEST_MSG)) |
365 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_SEND_COMPLETION_MSG)) |
366 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_RECV_COMPLETION_MSG)) |
367 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_CONGESTION_RELIEF_MSG));
368 114 : BQS_LOG_INFO("Enque group[%u] subscribe event, eventBitmap[%lu] deviceId[%u]", groupId, eventBitmap, deviceId);
369 114 : const int32_t ret = halEschedSubscribeEvent(deviceId, groupId, threadIndex, eventBitmap);
370 114 : if (ret != DRV_ERROR_NONE) {
371 0 : BQS_LOG_ERROR("halEschedSubscribeEvent failed, groupId[%u] eventBitmap[%lu] result[%d].",
372 : groupId, eventBitmap, ret);
373 0 : StopQueueSchedule();
374 0 : return;
375 : }
376 : // set max num for hccl event
377 114 : event_sched_grp_qos qos = {};
378 : const std::vector<uint32_t> eventList = {dgw::EVENT_RECV_REQUEST_MSG, dgw::EVENT_SEND_COMPLETION_MSG,
379 228 : dgw::EVENT_RECV_COMPLETION_MSG, EVENT_QUEUE_FULL_TO_NOT_FULL};
380 566 : for (const uint32_t eventId : eventList) {
381 452 : qos.maxNum = (eventId == static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL)) ?
382 : MAX_QOS_NUM_FOR_F2NF_EVENT : MAX_QOS_NUM_FOR_HCCL_EVENT;
383 452 : const auto drvRet = halEschedSetGrpEventQos(deviceId, groupId, static_cast<EVENT_ID>(eventId), &qos);
384 452 : if (drvRet != DRV_ERROR_NONE) {
385 0 : BQS_LOG_ERROR("Failed to call halEschedSetGrpEventQos, groupId[%u], qos.maxNum[%u], ret[%d].",
386 : groupId, qos.maxNum, static_cast<int32_t>(drvRet));
387 0 : StopQueueSchedule();
388 0 : return;
389 : }
390 : }
391 112 : LoopProcessEnqueueEvent(threadIndex, deviceId, groupId, index);
392 114 : }
393 :
394 159 : void QueueSchedule::BindAicpu(const uint32_t threadIndex, const uint32_t bindCpuIndex)
395 : {
396 159 : if (bqs::RunContext::HOST != bqs::GetRunContext()) {
397 8 : if (!hasAICPU_) {
398 : struct sched_param param;
399 4 : param.sched_priority = sched_get_priority_max(SCHED_FIFO);
400 4 : if (sched_setscheduler(0, SCHED_FIFO, ¶m) == -1) {
401 4 : BQS_LOG_ERROR("QueueSchedule sched_setscheduler failed, errno:%d, thread exit.", errno);
402 4 : StopQueueSchedule();
403 : }
404 : } else {
405 4 : if (!isZeroSizeAicpuNum_) {
406 2 : const int32_t status = BindCpuUtils::BindAicpu(bindCpuIndex);
407 2 : if (status != BQS_STATUS_OK) {
408 2 : BQS_LOG_ERROR(
409 : "QueueSchedule enqueue thread[%u] bind cpu[%u] failed, thread exit.",
410 : threadIndex, bindCpuIndex);
411 2 : StopQueueSchedule();
412 : }
413 : }
414 : }
415 : }
416 :
417 159 : if (BindCpuUtils::PostSem() != BQS_STATUS_OK) {
418 0 : BQS_LOG_ERROR("WaitPost failed");
419 : }
420 159 : }
421 :
422 154 : void QueueSchedule::CheckIfRecover(uint32_t &errCount, const char_t * const identity, const uint32_t threadIndex,
423 : const uint32_t groupId) const
424 : {
425 154 : if (errCount != 0U) {
426 1 : errCount = 0U;
427 1 : BQS_LOG_ERROR("halEschedWaitEvent %s event recover, threadIndex[%u] groupId[%u]",
428 : identity, threadIndex, groupId);
429 : }
430 154 : }
431 :
432 114 : void QueueSchedule::LoopProcessEnqueueEvent(const uint32_t threadIndex, const uint32_t deviceId,
433 : const uint32_t groupId, const uint32_t index)
434 : {
435 114 : QueueManager::GetInstance().NotifyInitSuccess(index);
436 115 : struct event_info event = {};
437 : // default wait timeout 2s
438 115 : constexpr int32_t waitTimeout = 2000;
439 115 : uint32_t errCount = 0U;
440 243 : while (running_) {
441 129 : StatisticManager::GetInstance().RefreshEnqueHeartBeat();
442 129 : const int32_t schedRet = halEschedWaitEvent(deviceId, groupId, threadIndex, waitTimeout, &event);
443 129 : if (schedRet == DRV_ERROR_NONE) {
444 111 : CheckIfRecover(errCount, "enqueue", threadIndex, groupId);
445 111 : StatisticManager::GetInstance().AwakenAdd();
446 111 : ProcessEvent(threadIndex, event, index);
447 18 : } else if (schedRet == DRV_ERROR_SCHED_WAIT_TIMEOUT) {
448 3 : CheckIfRecover(errCount, "enqueue", threadIndex, groupId);
449 3 : BQS_LOG_DEBUG("halEschedWaitEvent enqueue event timeout, groupId=%u, thread index:%u",
450 : groupId, threadIndex);
451 3 : continue;
452 15 : } else if (schedRet == DRV_ERROR_PARA_ERROR) {
453 1 : BQS_LOG_ERROR(
454 : "halEschedWaitEvent enqueue event failed, deviceId[%u] threadIndex[%u] groupId[%u] error[%d].",
455 : deviceId, threadIndex, groupId, schedRet);
456 1 : break;
457 : } else {
458 14 : if (errCount++ == 0U) {
459 1 : BQS_LOG_ERROR(
460 : "halEschedWaitEvent enqueue event failed, deviceId[%u] threadIndex[%u] groupId[%u] error[%d].",
461 : deviceId, threadIndex, groupId, schedRet);
462 : }
463 : }
464 : }
465 :
466 115 : BQS_LOG_INFO("QueueSchedule enqueue thread[%u] exit", threadIndex);
467 115 : }
468 :
469 45 : void QueueSchedule::F2NFThreadTask(const uint32_t threadIndex, const uint32_t bindCpuIndex, const uint32_t groupId)
470 : {
471 45 : BQS_LOG_INFO("Queue Schedule f2nf thread[%u] start.", threadIndex);
472 45 : const auto threadName = std::string(F2NF_THREAD_NAME_PREFIX).append(std::to_string(threadIndex));
473 45 : (void)pthread_setname_np(pthread_self(), threadName.c_str());
474 45 : BindAicpu(threadIndex, bindCpuIndex);
475 45 : const uint64_t eventBitmap =
476 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_RECV_REQUEST_MSG)) |
477 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_SEND_COMPLETION_MSG)) |
478 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_RECV_COMPLETION_MSG)) |
479 : static_cast<uint64_t>(1LU << static_cast<uint64_t>(dgw::EVENT_CONGESTION_RELIEF_MSG));
480 :
481 45 : BQS_LOG_INFO("F2NF group[%u] subscribe event, eventBitmap[%lu]", f2nfGroupId_, eventBitmap);
482 45 : const auto ret = halEschedSubscribeEvent(deviceId_, f2nfGroupId_, threadIndex, eventBitmap);
483 45 : if (ret != DRV_ERROR_NONE) {
484 0 : BQS_LOG_ERROR("halEschedSubscribeEvent failed, groupId[%u] eventBitmap[%lu] result[%d].",
485 : f2nfGroupId_, eventBitmap, static_cast<int32_t>(ret));
486 0 : StopQueueSchedule();
487 0 : return;
488 : }
489 :
490 : // set max num for hccl event
491 45 : event_sched_grp_qos qos = {};
492 : const std::vector<uint32_t> eventList = {dgw::EVENT_RECV_REQUEST_MSG, dgw::EVENT_SEND_COMPLETION_MSG,
493 90 : dgw::EVENT_RECV_COMPLETION_MSG};
494 180 : for (const uint32_t eventId : eventList) {
495 135 : qos.maxNum = MAX_QOS_NUM_FOR_HCCL_EVENT;
496 135 : const auto drvRet = halEschedSetGrpEventQos(deviceId_, f2nfGroupId_, static_cast<EVENT_ID>(eventId), &qos);
497 135 : if (drvRet != DRV_ERROR_NONE) {
498 0 : BQS_LOG_ERROR("Failed to call halEschedSetGrpEventQos, groupId[%u], qos.maxNum[%u], ret[%d].",
499 : f2nfGroupId_, qos.maxNum, static_cast<int32_t>(drvRet));
500 0 : StopQueueSchedule();
501 0 : return;
502 : }
503 : }
504 :
505 45 : struct event_info event = {};
506 : // default wait timeout 2s
507 45 : constexpr int32_t waitTimeout = 2000;
508 45 : uint32_t errCount = 0U;
509 99 : while (running_) {
510 55 : const int32_t schedRet = halEschedWaitEvent(deviceId_, groupId, threadIndex, waitTimeout, &event);
511 55 : if (schedRet == DRV_ERROR_NONE) {
512 40 : CheckIfRecover(errCount, "f2nf", threadIndex, groupId);
513 40 : (void)ProcessEvent(threadIndex, event, 0U);
514 15 : } else if (schedRet == DRV_ERROR_SCHED_WAIT_TIMEOUT) {
515 0 : CheckIfRecover(errCount, "f2nf", threadIndex, groupId);
516 0 : BQS_LOG_DEBUG("halEschedWaitEvent f2nf event timeout,thread index:%u", threadIndex);
517 0 : continue;
518 15 : } else if (schedRet == DRV_ERROR_PARA_ERROR) {
519 1 : BQS_LOG_ERROR(
520 : "halEschedWaitEvent f2nf event failed, deviceId[%u] threadIndex[%u] groupId[%u] error[%d].",
521 : deviceId_, threadIndex, groupId, schedRet);
522 1 : break;
523 : } else {
524 14 : if (errCount++ == 0U) {
525 1 : BQS_LOG_ERROR(
526 : "halEschedWaitEvent f2nf event failed, deviceId[%u] threadIndex[%u] groupId[%u] error[%d].",
527 : deviceId_, threadIndex, groupId, schedRet);
528 : }
529 : }
530 : }
531 45 : BQS_LOG_INFO("Queue Schedule f2nf thread[%u] exit", threadIndex);
532 45 : }
533 :
534 166 : BqsStatus QueueSchedule::ProcessEvent(const uint32_t threadIndex, event_info &event, const uint32_t index)
535 : {
536 166 : auto ret = dgw::FsmStatus::FSM_SUCCESS;
537 166 : const uint32_t eventId = event.comm.event_id;
538 166 : const uint32_t deviceId = (index == 0) ? deviceId_ : initQsParams_.deviceIdExtra;
539 :
540 166 : switch (eventId) {
541 0 : case static_cast<uint32_t>(EVENT_QUEUE_ENQUEUE): {
542 0 : BQS_LOG_INFO("the [%u]th thread[%u] recv enqueEvent", index, threadIndex);
543 0 : ProcessEnqueueEvent(threadIndex, event, index, false);
544 0 : break;
545 : }
546 1 : case static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL): {
547 1 : BQS_LOG_INFO("the [%u]th thread[%u] recv f2nfEvent", index, threadIndex);
548 1 : ProcessEnqueueEvent(threadIndex, event, index, true);
549 1 : break;
550 : }
551 3 : case dgw::EVENT_RECV_REQUEST_MSG: {
552 3 : ret = dgw::HcclProcess::GetInstance().ProcessRecvRequestEvent(event, deviceId, index);
553 3 : break;
554 : }
555 4 : case dgw::EVENT_SEND_COMPLETION_MSG: {
556 4 : ret = dgw::HcclProcess::GetInstance().ProcessSendCompletionEvent(event, deviceId, index);
557 4 : break;
558 : }
559 5 : case dgw::EVENT_RECV_COMPLETION_MSG: {
560 5 : ret = dgw::HcclProcess::GetInstance().ProcessRecvCompletionEvent(event, deviceId, index);
561 5 : break;
562 : }
563 1 : case dgw::EVENT_CONGESTION_RELIEF_MSG: {
564 1 : ret = dgw::HcclProcess::GetInstance().ProcessCongestionReliefEvent(event, deviceId, index);
565 1 : break;
566 : }
567 152 : default: {
568 152 : BQS_LOG_WARN("Unsupported event[%u].", eventId);
569 152 : ret = dgw::FsmStatus::FSM_FAILED;
570 152 : break;
571 : }
572 : }
573 :
574 166 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
575 152 : return BqsStatus::BQS_STATUS_INNER_ERROR;
576 : }
577 14 : return BqsStatus::BQS_STATUS_OK;
578 : }
579 :
580 4 : void QueueSchedule::DaemonThreadTask(const uint32_t index)
581 : {
582 4 : BQS_LOG_INFO("Queue Schedule Daemon thread start.");
583 4 : (void)pthread_setname_np(pthread_self(), DAEMON_THREAD_NAME_PREFIX);
584 4 : const uint32_t deviceId = (index == 0) ? deviceId_ : initQsParams_.deviceIdExtra;
585 4 : const uint32_t enqueGroupId = (index == 0) ? enqueGroupId_ : initQsParams_.enqueGroupIdExtra;
586 4 : BindCpuUtils::SetThreadFIFO(deviceId);
587 :
588 : // fixme: can run concurrancyly for different index
589 4 : std::unique_lock<std::mutex> daemonWaitLock(daemonWaitMtx_);
590 4 : uint64_t awakenTimes = 0UL;
591 : QueueSubscriber subscriber;
592 4 : subscriber.devId = deviceId;
593 4 : subscriber.spGrpId = 0;
594 4 : subscriber.pid = static_cast<int32_t>(getpid());
595 4 : subscriber.groupId = static_cast<int32_t>(enqueGroupId);
596 4 : while ((daemonWait_.wait_for(daemonWaitLock, std::chrono::milliseconds(reschedInterval_)) ==
597 4 : std::cv_status::timeout) && (running_)) {
598 0 : if (StatisticManager::GetInstance().GetEventScheduleStat() == 0U) {
599 : // no work to do
600 0 : continue;
601 : }
602 0 : const uint64_t newAwakenTimes = StatisticManager::GetInstance().GetAwakenTimes();
603 0 : if (awakenTimes == newAwakenTimes) {
604 0 : int32_t ret = halQueueCtrlEvent(&subscriber, QUE_PAUSE_EVENT);
605 0 : if (ret == DRV_ERROR_NONE) {
606 0 : DaemonEnqueueEvent(index);
607 0 : ret = halQueueCtrlEvent(&subscriber, QUE_RESUME_EVENT);
608 0 : if (ret != DRV_ERROR_NONE) {
609 0 : BQS_LOG_ERROR(
610 : "halQueueCtrlEvent QUE_RESUME_EVENT failed, deviceId_[%u] enqueGroupId_[%u] ret[%d].",
611 : deviceId, enqueGroupId, ret);
612 : }
613 : } else {
614 0 : BQS_LOG_RUN_WARN("halQueueCtrlEvent QUE_PAUSE_EVENT failed, deviceId_[%u] enqueGroupId_[%u] ret[%d].",
615 : deviceId, enqueGroupId, ret);
616 : }
617 : } else {
618 0 : awakenTimes = newAwakenTimes;
619 : }
620 : }
621 4 : BQS_LOG_INFO("Queue Schedule Daemon thread exit");
622 4 : }
623 :
624 3 : void QueueSchedule::ProcessEnqueueEvent(const uint32_t threadIndex, const event_info &event, const uint32_t index,
625 : const bool procF2NF)
626 : {
627 3 : auto &queueEventAtomicFlag = (index == 0U) ? queueEventAtomicFlag_ : queueEventAtomicFlagExtra_;
628 : // if other thread is working do nothing; if not set work flag.
629 3 : if (!queueEventAtomicFlag.test_and_set()) {
630 1 : ProfileManager &profileManager = ProfileManager::GetInstance(index);
631 1 : const uint64_t eventBegin = profileManager.GetCpuTick();
632 1 : const uint64_t schedDelay = static_cast<uint64_t>(event.comm.sched_timestamp - event.comm.submit_timestamp);
633 1 : const uint64_t schedTimes = StatisticManager::GetInstance().EventScheduleStat();
634 1 : profileManager.InitMaker(schedTimes, schedDelay);
635 :
636 : // handle relation queue
637 1 : const bool procRelation = QueueManager::GetInstance().HandleRelationEvent(index);
638 1 : const uint64_t f2NFBegin = profileManager.GetCpuTick();
639 1 : profileManager.SetRelationCost(f2NFBegin - eventBegin);
640 :
641 : // handle full to not full queue
642 1 : bool hasF2NF = QueueManager::GetInstance().HandleFullToNotFullEvent(index);
643 1 : profileManager.Setf2NFCost(profileManager.GetCpuTick() - f2NFBegin);
644 1 : hasF2NF |= procF2NF;
645 :
646 : // handle schedule data
647 1 : const bool procAsynMemBuff = QueueManager::GetInstance().HandleAsynMemBuffEvent(index);
648 1 : const uint64_t scheduleBegin = profileManager.GetCpuTick();
649 1 : ScheduleDataBuffAll(!(procRelation || hasF2NF || procAsynMemBuff), index);
650 2 : StatisticManager::GetInstance().UpdateScheuleStatistic(profileManager.GetTimeCost(schedDelay),
651 1 : profileManager.GetTimeCost(profileManager.GetCpuTick() - scheduleBegin));
652 1 : profileManager.TryMarker(eventBegin);
653 : // clear event work flag
654 : queueEventAtomicFlag.clear();
655 1 : return;
656 : }
657 :
658 2 : if (procF2NF) {
659 1 : ProcessFullToNotFullEvent(index);
660 1 : return;
661 : }
662 :
663 1 : bqs::StatisticManager::GetInstance().EnqueueEventFalseAwakenStat();
664 1 : BQS_LOG_DEBUG("Thread[%u] can't work as other thread is working.", threadIndex);
665 : }
666 :
667 1 : void QueueSchedule::ProcessFullToNotFullEvent(const uint32_t index)
668 : {
669 1 : bqs::StatisticManager::GetInstance().F2nfEventStat();
670 1 : (void)QueueManager::GetInstance().EnqueueFullToNotFullEvent(index);
671 1 : }
672 :
673 1 : void QueueSchedule::DaemonEnqueueEvent(const uint32_t index)
674 : {
675 1 : auto &queueEventAtomicFlag = (index == 0U) ? queueEventAtomicFlag_ : queueEventAtomicFlagExtra_;
676 : // if other thread is working do nothing; if not set work flag.
677 1 : if (!queueEventAtomicFlag.test_and_set()) {
678 1 : StatisticManager::GetInstance().DaemonEventScheduleStat();
679 :
680 : // handle relation queue
681 1 : const bool procRelation = QueueManager::GetInstance().HandleRelationEvent(index);
682 :
683 : // handle schedule data
684 1 : ScheduleDataBuffAll(!procRelation, index);
685 :
686 : // clear event work flag
687 : queueEventAtomicFlag.clear();
688 : } else {
689 0 : BQS_LOG_WARN("Daemon thread can't work as other thread is working, may event error.");
690 : }
691 1 : }
692 :
693 29 : void QueueSchedule::DynamicSchedule(const uint32_t index) const
694 : {
695 29 : const auto &dynamicCfgKeys = dgw::ScheduleConfig::GetInstance().GetSchedKeys();
696 29 : BQS_LOG_DEBUG("dynamicCfgKeys size is %zu", dynamicCfgKeys.size());
697 29 : if (dynamicCfgKeys.empty()) {
698 28 : return;
699 : }
700 1 : uint32_t dynamicScheduleCount = 0U;
701 2 : for (const auto key: dynamicCfgKeys) {
702 1 : if (dgw::ScheduleConfig::GetInstance().IsStopped(key)) {
703 0 : BQS_LOG_INFO("key[%u] is stopped, then skip", key);
704 0 : continue;
705 : }
706 2 : while (dynamicScheduleCount++ < 100U) {
707 2 : std::vector<dgw::DynamicSchedMgr::ResponseInfo> responses;
708 2 : const auto getResponseRet = dgw::DynamicSchedMgr::GetInstance(index).GetResponse(key, responses);
709 2 : if ((getResponseRet != dgw::FsmStatus::FSM_SUCCESS) || (responses.size() == 0U)) {
710 1 : BQS_LOG_DEBUG("Can't get response, ret is %d", static_cast<int32_t>(getResponseRet));
711 1 : break;
712 : }
713 1 : DynamicScheduleByResponse(key, index, responses);
714 2 : }
715 : }
716 1 : BQS_LOG_DEBUG("finish DynamicSchedule, dynamicScheduleCount is %u", dynamicScheduleCount);
717 : }
718 :
719 28 : void QueueSchedule::ScheduleDataBuffAll(const bool dataEnqueue, const uint32_t index) const
720 : {
721 28 : BQS_LOG_INFO("the [%u]th thread ScheduleDataBuffAll.", index);
722 28 : bool hasDequeueFlag = false;
723 28 : const auto &orderedSubscribeQueues = (index == 0U) ?
724 28 : BindRelation::GetInstance().GetOrderedSubscribeQueueId() :
725 0 : BindRelation::GetInstance().GetOrderedSubscribeQueueIdExtra();
726 : const auto &srcToDstRelation = (index == 0U) ?
727 28 : BindRelation::GetInstance().GetSrcToDstRelation() :
728 28 : BindRelation::GetInstance().GetSrcToDstExtraRelation();
729 :
730 28 : ProfileManager::GetInstance(index).SetSrcQueueNum(static_cast<uint32_t>(orderedSubscribeQueues.size()));
731 28 : dgw::InnerMessage msg;
732 28 : msg.msgType = dgw::InnerMsgType::INNER_MSG_PUSH;
733 28 : BindRelation::GetInstance().ClearAbnormalEntityInfo(index);
734 32 : if (dgw::EntityManager::Instance(index).IsExistFullEntity() ||
735 4 : dgw::EntityManager::Instance(index).IsExistAsyncMemEntity()) {
736 47 : for (const auto &src : orderedSubscribeQueues) {
737 24 : if (dgw::ScheduleConfig::GetInstance().IsStopped(src.GetSchedCfgKey())) {
738 0 : BQS_LOG_INFO("Skip schedule src[%s] for it has been stopped.", src.ToString().c_str());
739 1 : continue;
740 : }
741 24 : const auto iter = srcToDstRelation.find(src);
742 24 : if (iter == srcToDstRelation.end()) {
743 1 : BQS_LOG_WARN("Can't find dst queues for queue[%u].", src.GetId());
744 1 : continue;
745 : }
746 46 : for (auto &dst : iter->second) {
747 : // process full state for dst entity
748 23 : if (ProcessDstEntity(dst, index) == dgw::FsmStatus::FSM_ERROR) {
749 0 : BQS_LOG_ERROR("Skip scheduler for routes maybe has been modified");
750 0 : break;
751 : };
752 : }
753 :
754 23 : const auto &srcEntity = src.GetEntity();
755 23 : if (srcEntity->ProcessMessage(msg) == dgw::FsmStatus::FSM_ERROR) {
756 1 : BQS_LOG_ERROR("skip scheduler for routes maybe have been modified");
757 1 : break;
758 : }
759 22 : hasDequeueFlag |= (srcEntity->GetScheduleCount() > 0UL);
760 : }
761 : } else {
762 6 : for (const auto &src : orderedSubscribeQueues) {
763 3 : if (dgw::ScheduleConfig::GetInstance().IsStopped(src.GetSchedCfgKey())) {
764 0 : BQS_LOG_INFO("Skip schedule src[%s] for it has been stopped.", src.ToString().c_str());
765 0 : continue;
766 : }
767 3 : const auto &srcEntity = src.GetEntity();
768 3 : if (srcEntity->ProcessMessage(msg) == dgw::FsmStatus::FSM_ERROR) {
769 1 : BQS_LOG_ERROR("skip scheduler for routes maybe have been modified");
770 1 : break;
771 : }
772 2 : hasDequeueFlag |= (srcEntity->GetScheduleCount() > 0UL);
773 : }
774 : }
775 :
776 28 : DynamicSchedule(index);
777 :
778 : // supply recv request event
779 28 : if (hasDequeueFlag && (!aicpuFeatureDisableRecvRequestEvent_)) {
780 12 : (void)dgw::EntityManager::Instance(index).SupplyRecvRequestEvent();
781 : }
782 :
783 28 : if ((!hasDequeueFlag) && dataEnqueue) {
784 14 : StatisticManager::GetInstance().AddScheduleEmpty();
785 : }
786 28 : BindRelation::GetInstance().UpdateRelation(index);
787 28 : }
788 :
789 23 : dgw::FsmStatus QueueSchedule::ProcessDstEntity(const EntityInfo &entity, const uint32_t index) const
790 : {
791 23 : const auto dstEntity = entity.GetEntity();
792 23 : if (dstEntity == nullptr) {
793 0 : BQS_LOG_ERROR("Get entity ptr for entity[%s] failed.", entity.ToString().c_str());
794 0 : return dgw::FsmStatus::FSM_FAILED;
795 : }
796 : // process full for queue or tag
797 23 : if (entity.GetType() != dgw::EntityType::ENTITY_GROUP) {
798 22 : BQS_LOG_DEBUG("Process dst entity, id[%u], type[%s].",
799 : dstEntity->GetId(), dstEntity->GetTypeDesc().c_str());
800 22 : dgw::InnerMessage msg;
801 22 : msg.msgType = dstEntity->GetCurState() == dgw::FsmState::FSM_FULL_STATE ? dgw::InnerMsgType::INNER_MSG_F2NF :
802 : dgw::InnerMsgType::INNER_MSG_PUSH;
803 22 : return dstEntity->ProcessMessage(msg);
804 : }
805 : // process full for group
806 1 : auto &entitiesInGroup = dgw::EntityManager::Instance(index).GetEntitiesInGroup(entity.GetId());
807 3 : for (auto &entityInGroup : entitiesInGroup) {
808 2 : BQS_LOG_DEBUG("Process dst entity in group[%u], id[%u], type[%s].",
809 : dstEntity->GetId(), entityInGroup->GetId(), entityInGroup->GetTypeDesc().c_str());
810 2 : dgw::InnerMessage msg;
811 2 : msg.msgType = entityInGroup->GetCurState() == dgw::FsmState::FSM_FULL_STATE ?
812 : dgw::InnerMsgType::INNER_MSG_F2NF : dgw::InnerMsgType::INNER_MSG_PUSH;
813 2 : (void) entityInGroup->ProcessMessage(msg);
814 : }
815 1 : return dgw::FsmStatus::FSM_SUCCESS;
816 23 : }
817 :
818 108 : QueueSchedule::~QueueSchedule()
819 : {
820 108 : running_ = false;
821 108 : daemonWait_.notify_all();
822 270 : for (auto &worker : workThreads_) {
823 162 : if (worker.joinable()) {
824 0 : worker.join();
825 : }
826 : }
827 108 : }
828 :
829 158 : void QueueSchedule::WaitForStop()
830 : {
831 158 : BQS_LOG_RUN_INFO("WaitForStop begin");
832 158 : RouterServer::GetInstance().Destroy();
833 158 : StopQueueSchedule();
834 482 : for (auto &worker : workThreads_) {
835 324 : if (worker.joinable()) {
836 162 : worker.join();
837 : }
838 : }
839 158 : BQS_LOG_INFO("WaitForStop end");
840 158 : }
841 :
842 : /* *
843 : * init drv event scheduler.
844 : * @return BQS_STATUS_OK: success, other: error
845 : */
846 19 : BqsStatus QueueSchedule::InitDrvSchedModule(const uint32_t deviceId, const uint32_t enqueGroupId,
847 : const uint32_t f2nfGroupId) const
848 : {
849 19 : BQS_LOG_INFO("Attach device[%u] to drv scheduler", deviceId);
850 : (void)f2nfGroupId;
851 19 : int32_t ret = halEschedAttachDevice(deviceId);
852 19 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_PROCESS_REPEAT_ADD)) {
853 1 : BQS_LOG_ERROR("Failed to attach device[%u] for eSched, result[%d].", deviceId, ret);
854 1 : return BQS_STATUS_DRIVER_ERROR;
855 : }
856 :
857 22 : for (uint32_t idx = 0; idx < initQsParams_.devIdVec.size(); idx++) {
858 4 : uint32_t currDeviceId = initQsParams_.devIdVec[idx];
859 4 : ret = halEschedAttachDevice(currDeviceId);
860 4 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_PROCESS_REPEAT_ADD)) {
861 0 : BQS_LOG_ERROR("Failed to attach device[%u] for eSched, result[%d].", currDeviceId, ret);
862 0 : return BQS_STATUS_DRIVER_ERROR;
863 : }
864 :
865 4 : if (deviceId_ == currDeviceId) {
866 0 : continue;
867 : }
868 : QueueSetInputPara inPutParam;
869 4 : (void)halQueueSet(currDeviceId, QUEUE_ENABLE_LOCAL_QUEUE, &inPutParam);
870 :
871 4 : ret = halQueueInit(currDeviceId);
872 4 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_REPEATED_INIT)) {
873 0 : BQS_LOG_ERROR("host flow halQueueInit error, ret=[%d]", static_cast<int32_t>(ret));
874 0 : return BQS_STATUS_DRIVER_ERROR;
875 : }
876 : }
877 :
878 : // set pid priority
879 24 : const bool setPidPriorityFlag = (bqs::GetRunContext() == bqs::RunContext::HOST) ?
880 6 : true : aicpuFeatureSetPidPriority_;
881 18 : if (setPidPriorityFlag) {
882 12 : (void)halEschedSetPidPriority(deviceId, PRIORITY_LEVEL0);
883 : }
884 :
885 18 : GROUP_TYPE enqueGrpType = GRP_TYPE_BIND_CP_CPU;
886 18 : GROUP_TYPE f2nfGrpType = GRP_TYPE_BIND_CP_CPU;
887 18 : if (hasAICPU_ && !isZeroSizeAicpuNum_ && (bqs::RunContext::HOST != bqs::GetRunContext())) {
888 1 : enqueGrpType = GRP_TYPE_BIND_DP_CPU;
889 1 : f2nfGrpType = GRP_TYPE_BIND_DP_CPU;
890 : }
891 18 : BQS_LOG_INFO("Create enqueGroup[%u] type[%d] on device[%u].", enqueGroupId, enqueGrpType, deviceId);
892 18 : ret = halEschedCreateGrp(deviceId, enqueGroupId, enqueGrpType);
893 18 : if (ret != DRV_ERROR_NONE) {
894 1 : (void)halEschedDettachDevice(deviceId);
895 1 : BQS_LOG_ERROR("Failed to create enqueGroup, groupId[%u] result[%d].", enqueGroupId, ret);
896 1 : return BQS_STATUS_DRIVER_ERROR;
897 : }
898 :
899 17 : if (!initQsParams_.numaFlag) {
900 9 : BQS_LOG_INFO("Create f2nfGroup[%u], type[%d]", f2nfGroupId_, f2nfGrpType);
901 9 : ret = halEschedCreateGrp(deviceId_, f2nfGroupId_, f2nfGrpType);
902 9 : if (ret != DRV_ERROR_NONE) {
903 0 : (void)halEschedDettachDevice(deviceId_);
904 0 : BQS_LOG_ERROR("Failed to create f2nfGroup, groupId[%u] result[%d].", f2nfGroupId_, ret);
905 0 : return BQS_STATUS_DRIVER_ERROR;
906 : }
907 : }
908 17 : return BQS_STATUS_OK;
909 : }
910 :
911 1 : void QueueSchedule::ReportAbnormal() const
912 : {
913 1 : BQS_LOG_ERROR("Enqueue thread has missed heartbeat for %u seconds", abnormalInterval_);
914 1 : if ((bqs::GetRunContext() != bqs::RunContext::HOST) &&
915 2 : (initQsParams_.starter != bqs::QsStartType::START_BY_DEPLOYER) &&
916 1 : (initQsParams_.runMode != QueueSchedulerRunMode::MULTI_THREAD)) {
917 1 : const int32_t ret = TsdDestroy(deviceId_, TSD_QS, initQsParams_.pid, initQsParams_.vfId);
918 1 : if (ret != 0) {
919 0 : BQS_LOG_ERROR("dev[%u] send abnormal msg to tsdaemon failed, ret[%d]", deviceId_, ret);
920 : }
921 : }
922 1 : }
923 : } // namespace bqs
|
