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