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 "aicpusd_worker.h"
12 :
13 : #include <csignal>
14 : #include <cstring>
15 : #include <cerrno>
16 : #include <sys/wait.h>
17 : #include <algorithm>
18 :
19 : #include "tsd.h"
20 : #include "aicpusd_status.h"
21 : #include "aicpusd_util.h"
22 : #include "aicpusd_drv_manager.h"
23 : #include "aicpusd_monitor.h"
24 : #include "aicpusd_event_manager.h"
25 : #include "aicpusd_context.h"
26 : #include "aicpu_context.h"
27 : #include "aicpusd_proc_mgr_sys_operator_agent.h"
28 : #include "aicpusd_hal_interface_ref.h"
29 : #include "aicpusd_so_manager.h"
30 : #include "aicpu_pulse.h"
31 : #include "aicpusd_feature_ctrl.h"
32 : #include "aicpusd_message_queue.h"
33 :
34 : namespace {
35 : // user event id starts from EVENT_USR_START(48) to EVENT_USR_END(63), we should make sure that eventIds in
36 : // one process should not conflict, considering the first 3 user_event_id has been used by qs, so here we start by
37 : // offset 8
38 : constexpr uint64_t EVENT_PROXY_MSG = static_cast<uint64_t>(EVENT_ID::EVENT_USR_START) + 8U;
39 :
40 : constexpr uint64_t CP_EVENT_MASK =
41 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_RANDOM_KERNEL)) |
42 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_DVPP_MSG)) |
43 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_FR_MSG)) |
44 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_TS_HWTS_KERNEL)) |
45 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_TS_HWTS_KERNEL)) |
46 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_AICPU_MSG)) |
47 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_TS_CTRL_MSG)) |
48 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_QUEUE_EMPTY_TO_NOT_EMPTY)) |
49 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_QUEUE_FULL_TO_NOT_FULL)) |
50 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_TDT_ENQUEUE)) |
51 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_ACPU_MSG_TYPE1)) |
52 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_DVPP_MPI_MSG)) |
53 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_SPLIT_KERNEL)) |
54 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_CDQ_MSG)) |
55 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_QUEUE_ENQUEUE)) |
56 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_QS_MSG)) |
57 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_DRV_MSG)) |
58 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << static_cast<uint64_t>(EVENT_FFTS_PLUS_MSG)) |
59 : static_cast<uint64_t>(static_cast<uint64_t>(1U) << EVENT_PROXY_MSG);
60 : constexpr const uint32_t SLEEP_USECS = 50000U;
61 : // When there is no AICPU core, the AICPU schedule starts this number of work threads
62 : // and binds them to the largest CTRLCPU cores.
63 : constexpr size_t NO_AICPU_WORKER_NUM = 2UL;
64 : }
65 :
66 : namespace AicpuSchedule {
67 101 : ThreadPool &ThreadPool::Instance()
68 : {
69 101 : static ThreadPool threadPoolInstance;
70 101 : return threadPoolInstance;
71 : }
72 :
73 36 : ThreadPool::ThreadPool() : semInitedNum_(0UL)
74 : {
75 36 : aicpusd_run_info("ThreadPool");
76 36 : }
77 :
78 36 : ThreadPool::~ThreadPool()
79 : {
80 36 : ClearPulseNotifyFunc();
81 36 : Clear();
82 36 : }
83 :
84 48 : void ThreadPool::Clear()
85 : {
86 48 : AicpuSchedule::AicpuEventManager::GetInstance().SetRunningFlag(false);
87 48 : WaitForStop();
88 71 : for (auto &sem : sems_) {
89 23 : (void)sem_destroy(&sem);
90 : }
91 48 : semInitedNum_ = 0UL;
92 48 : threadStatusList_.clear();
93 48 : threadIdLists_.clear();
94 48 : }
95 :
96 38 : size_t ThreadPool::GetWorkerNum()
97 : {
98 38 : const size_t aicpuNum = static_cast<size_t>(AicpuDrvManager::GetInstance().GetAicpuNum());
99 : // When there is no AICPU core, the AICPU schedule still starts NO_AICPU_WORKER_NUM work threads.
100 38 : return (aicpuNum == 0UL) ? NO_AICPU_WORKER_NUM : aicpuNum;
101 : }
102 :
103 12 : int32_t ThreadPool::CreateWorker(const AicpuSchedMode schedMode)
104 : {
105 12 : Clear();
106 12 : AicpuSchedule::AicpuEventManager::GetInstance().SetRunningFlag(true);
107 12 : schedMode_ = schedMode;
108 12 : size_t aicpuNum = GetWorkerNum();
109 12 : const std::vector<uint32_t> deviceVec = AicpuDrvManager::GetInstance().GetDeviceList();
110 12 : if (AicpuDrvManager::GetInstance().GetAicpuNum() == 0UL) {
111 2 : aicpusd_run_info("aicpu total num=[0], create [%zu] aicpu workers", aicpuNum);
112 2 : hasAicpu_ = false;
113 : }
114 12 : sems_ = std::vector<sem_t>(static_cast<size_t>(aicpuNum));
115 34 : for (size_t i = 0UL; i < aicpuNum; ++i) {
116 23 : const int32_t semInitRet = sem_init(&(sems_[i]), 0, 0U);
117 23 : if (semInitRet == -1) {
118 1 : aicpusd_err("sem[%zu] init failed, %s", i, strerror(errno));
119 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
120 : }
121 22 : semInitedNum_ = i + 1UL;
122 : }
123 11 : threadStatusList_ = std::vector<ThreadStatus>(static_cast<size_t>(aicpuNum), ThreadStatus::THREAD_INIT);
124 11 : threadIdLists_ = std::vector<pid_t>(static_cast<size_t>(aicpuNum), 0);
125 :
126 11 : sighandler_t const oldHandler = signal(SIGCHLD, SIG_DFL);
127 11 : aicpusd_info("set SIGCHLD to %d, old sighandler[%d], errno[%d]", SIG_DFL, oldHandler, errno);
128 :
129 : // When there is no AICPU core, all work threads run on the same device (deviceVec[0]),
130 : // so use aicpuNum as the per-device count to keep deviceVecInx at 0 for every worker.
131 11 : const size_t aicpuNumPerDev = hasAicpu_ ? AicpuDrvManager::GetInstance().GetAicpuNumPerDevice() : aicpuNum;
132 11 : if (aicpuNumPerDev == 0) {
133 1 : aicpusd_err("aicpu Number error, %lu", aicpuNumPerDev);
134 1 : (void)signal(SIGCHLD, oldHandler);
135 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
136 : }
137 :
138 10 : int32_t ret = AICPU_SCHEDULE_OK;
139 30 : for (size_t i = 0UL; i < aicpuNum; ++i) {
140 : // aicpuNumPerDev is not 0
141 21 : const size_t deviceVecInx = i / aicpuNumPerDev;
142 21 : ret = CreateOneWorker(i, deviceVec[deviceVecInx]);
143 21 : if (ret != AICPU_SCHEDULE_OK) {
144 1 : (void)signal(SIGCHLD, oldHandler);
145 1 : return ret;
146 : }
147 : }
148 :
149 21 : for (size_t i = 0UL; i < aicpuNum; i++) {
150 18 : const int32_t semWaitRet = sem_wait(&(sems_[i]));
151 18 : if (semWaitRet == -1) {
152 1 : (void)signal(SIGCHLD, oldHandler);
153 1 : aicpusd_err("sem[%zu] wait failed, %s", i, strerror(errno));
154 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
155 : }
156 17 : if (threadStatusList_[i] != ThreadStatus::THREAD_RUNNING) {
157 5 : (void)signal(SIGCHLD, oldHandler);
158 5 : aicpusd_err("create thread[%zu] failed", i);
159 5 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
160 : }
161 : }
162 3 : aicpusd_info("set SIGCHLD to old sighandler[%d]", oldHandler);
163 3 : (void)signal(SIGCHLD, oldHandler);
164 :
165 3 : ret = AicpuSchedule::AicpuMonitor::GetInstance().Run();
166 3 : if (ret != AICPU_SCHEDULE_OK) {
167 0 : aicpusd_err("aicpu monitor run failed, ret[%d]", ret);
168 0 : return ret;
169 : }
170 :
171 3 : return AICPU_SCHEDULE_OK;
172 12 : }
173 :
174 13 : int32_t ThreadPool::CreateOneWorker(const size_t threadIndex,
175 : const uint32_t deviceId)
176 : {
177 : try {
178 13 : aicpusd_info("CreateOneWorker device[%u]:thread[%zu] started.", deviceId, threadIndex);
179 13 : std::thread th(&ThreadPool::Work, threadIndex, deviceId, schedMode_);
180 13 : workers_.emplace_back(std::move(th));
181 13 : } catch (std::exception &threadException) {
182 0 : aicpusd_err("create aicpu worker[%zu] failed, %s", threadIndex, threadException.what());
183 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
184 0 : }
185 :
186 13 : return AICPU_SCHEDULE_OK;
187 : }
188 :
189 69 : void ThreadPool::WaitForStop()
190 : {
191 69 : aicpusd_run_info("Wait for stop begin.");
192 82 : for (auto &worker : workers_) {
193 13 : if (worker.joinable()) {
194 13 : worker.join();
195 : }
196 : }
197 69 : workers_.clear();
198 69 : aicpusd_run_info("Wait for stop end.");
199 69 : }
200 :
201 17 : void ThreadPool::Work(const size_t threadIndex, const uint32_t deviceId, const AicpuSchedMode schedMode)
202 : {
203 17 : aicpusd_info("Aicpu device[%u]:thread[%zu] started.", deviceId, threadIndex);
204 : aicpu::aicpuContext_t context;
205 17 : context.tsId = 0U;
206 17 : context.hostPid = AicpuDrvManager::GetInstance().GetHostPid();
207 17 : context.vfId = AicpuDrvManager::GetInstance().GetVfId();
208 17 : context.deviceId = deviceId;
209 17 : aicpu::SetUniqueVfId(AicpuDrvManager::GetInstance().GetUniqueVfId());
210 17 : if (aicpu::aicpuSetContext(&context) != aicpu::AICPU_ERROR_NONE) {
211 0 : aicpusd_err("Set aicpu context failed, deviceId[%u], thread[%zu].", deviceId, threadIndex);
212 0 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
213 3 : return;
214 : }
215 :
216 17 : DeployContext deployCtx = DeployContext::DEVICE;
217 17 : const StatusCode ctxRet = GetAicpuDeployContext(deployCtx);
218 17 : if (ctxRet != AICPU_SCHEDULE_OK) {
219 0 : aicpusd_err("Get current deploy ctx failed.");
220 0 : return;
221 : }
222 :
223 17 : if (deployCtx == DeployContext::DEVICE) {
224 17 : if (AicpuSchedule::ThreadPool::Instance().SetAffinity(threadIndex, deviceId) != AICPU_SCHEDULE_OK) {
225 2 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
226 2 : return;
227 : }
228 : } else {
229 0 : AicpuSchedule::ThreadPool::Instance().SetThreadStatus(threadIndex, ThreadStatus::THREAD_RUNNING);
230 : }
231 15 : AicpuSchedule::ThreadPool::Instance().SetThreadIdRelation(threadIndex, static_cast<pid_t>(GetTid()));
232 :
233 15 : const int32_t ret = (schedMode == SCHED_MODE_MSGQ) ? InitMessageQueueWorker(threadIndex) :
234 13 : InitInterruptWorker(deviceId, threadIndex);
235 15 : if (ret != AICPU_SCHEDULE_OK) {
236 1 : aicpusd_err("Init work for sched mode failed, mode=%u", schedMode);
237 1 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
238 1 : return;
239 : }
240 :
241 14 : (void)aicpu::SetAicpuThreadIndex(static_cast<uint32_t>(threadIndex));
242 : // virtual device：给dvpp传递deviceId
243 14 : if (FeatureCtrl::IsVfModeCheckedByDeviceId(deviceId)) {
244 0 : AicpuSoManager::GetInstance().SetDeviceIdToDvpp(deviceId);
245 : }
246 :
247 : // should not add any process between PostSem and LoopProcess
248 14 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
249 14 : AicpuSchedule::AicpuEventManager::GetInstance().LoopProcess(static_cast<uint32_t>(threadIndex));
250 :
251 14 : aicpusd_info("Aicpu device[%u]:thread[%u] stopped.", deviceId, threadIndex);
252 : }
253 :
254 14 : int32_t ThreadPool::InitInterruptWorker(const uint32_t deviceId, const size_t threadIndex)
255 : {
256 14 : const int32_t ret = halEschedSubscribeEvent(deviceId, CP_DEFAULT_GROUP_ID,
257 14 : static_cast<uint32_t>(threadIndex), CP_EVENT_MASK);
258 14 : if (ret != DRV_ERROR_NONE) {
259 1 : aicpusd_err("Subscribe event failed, deviceId[%u], groupId[%u], threadIndex[%zu] "
260 : "eventBitmap[%llu].", deviceId, CP_DEFAULT_GROUP_ID, threadIndex, CP_EVENT_MASK);
261 1 : return ret;
262 : }
263 :
264 13 : aicpusd_info("halEschedSubscribeEvent success, deviceId[%u], groupId[%u], threadIndex[%zu] eventBitmap[%llu].",
265 : deviceId, CP_DEFAULT_GROUP_ID, threadIndex, CP_EVENT_MASK);
266 :
267 : /**
268 : * In some multi-thread scenarios, the queue for esched may not be created before the RTS delivers
269 : * the AICPU task. As a result, the abnormal status will be send to RTS. Therefore, need to invoke
270 : * the halEschedWaitEvent in advance to create a waiting queue for esched.
271 : */
272 13 : (void)AicpuSchedule::AicpuEventManager::GetInstance().DoOnce(static_cast<uint32_t>(threadIndex), deviceId, 0);
273 :
274 13 : return AICPU_SCHEDULE_OK;
275 : }
276 :
277 2 : int32_t ThreadPool::InitMessageQueueWorker(const size_t threadIndex)
278 : {
279 2 : return MessageQueue::GetInstance().InitMessageQueueForThread(threadIndex);
280 : }
281 :
282 19 : int32_t ThreadPool::WriteTidForAffinity(const size_t threadIndex)
283 : {
284 19 : if (threadIndex >= threadStatusList_.size()) {
285 2 : aicpusd_err("threadIndex[%zu], out of rank[0, %zu]",
286 : threadIndex, threadStatusList_.size());
287 2 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
288 : }
289 :
290 34 : std::string command = "sudo /var/add_aicpu_tid_to_tasks.sh";
291 17 : std::string pathStr = "/var/add_aicpu_tid_to_tasks.sh";
292 17 : if (access(pathStr.c_str(), F_OK) != 0) {
293 15 : aicpusd_info("Not find add_aicpu_tid_to_tasks.sh.");
294 15 : return AICPU_SCHEDULE_OK;
295 : }
296 2 : command = command + " " + std::to_string(GetTid());
297 :
298 : // 使用system命令会对父进程进行拷贝，浪费了系统资源。在esl等环境中还会存在由于资源较少无法fork导致system卡住的问题.
299 : // 使用vfork替换system命令，由于与父进程共享资源，因此可解决资源浪费/卡住的问题.
300 2 : const int32_t ret = AicpuUtil::ExecuteCmd(command);
301 2 : if (ret != 0) {
302 1 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_EXIT;
303 1 : aicpusd_err("write tid[%llu] to /sys/fs/cgroup/cpuset/AICPU/tasks failed, ret[%d], strerror[%s]",
304 : GetTid(), ret, strerror(errno));
305 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
306 : }
307 :
308 1 : return AICPU_SCHEDULE_OK;
309 17 : }
310 :
311 20 : int32_t ThreadPool::AddPidToTask(const size_t threadIndex)
312 : {
313 20 : if (FeatureCtrl::IsBindPidByHal()) {
314 2 : if (&halBindCgroup != nullptr) {
315 2 : aicpusd_info("Bind pid by hal index:%zu.", threadIndex);
316 2 : const drvError_t drvRet = halBindCgroup(BIND_AICPU_CGROUP);
317 2 : if (drvRet != DRV_ERROR_NONE) {
318 1 : aicpusd_err("halBindCgroup failed, ret[%d]", drvRet);
319 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
320 : }
321 1 : aicpusd_info("halBindCgroup success");
322 : }
323 : } else {
324 18 : aicpusd_run_info("AddPidToTask by WriteTidForAffinity");
325 18 : const auto ret = WriteTidForAffinity(threadIndex);
326 18 : if (ret != static_cast<int32_t>(AICPU_SCHEDULE_OK)) {
327 2 : aicpusd_err("WriteTidForAffinity failed, ret[%d]", ret);
328 2 : return static_cast<int32_t>(AICPU_SCHEDULE_ERROR_INIT_FAILED);
329 : }
330 16 : aicpusd_info("WriteTidForAffinity success");
331 : }
332 17 : return AICPU_SCHEDULE_OK;
333 : }
334 :
335 11 : uint32_t ThreadPool::GetNoAicpuCcpuPhysIndex(const size_t threadIndex, const uint32_t deviceId) const
336 : {
337 11 : const uint32_t ccpuNum = AicpuDrvManager::GetInstance().GetCcpuNum();
338 11 : if (ccpuNum == 0U) {
339 1 : aicpusd_err("no ctrlcpu core available for no-aicpu worker[%zu]", threadIndex);
340 1 : return INVALID_AICPU_ID;
341 : }
342 : // No AICPU core: bind work threads to the largest CTRLCPU cores. ccpuIdVec_ is sorted in
343 : // ascending order, so the last element is the largest core. Worker 0 takes the largest core,
344 : // worker 1 the second largest, and so on. When there are fewer cores than workers, the extra
345 : // workers fall back to the smallest core.
346 10 : uint32_t ccpuLogIndex = 0U;
347 10 : if (static_cast<size_t>(ccpuNum) > threadIndex) {
348 9 : ccpuLogIndex = ccpuNum - 1U - static_cast<uint32_t>(threadIndex);
349 : }
350 10 : const uint32_t physIndex = AicpuDrvManager::GetInstance().GetCcpuPhysIndex(ccpuLogIndex, deviceId);
351 10 : aicpusd_info("no aicpu worker[%zu] bind to ctrlcpu logIndex[%u], physIndex[%u]",
352 : threadIndex, ccpuLogIndex, physIndex);
353 10 : return physIndex;
354 : }
355 :
356 22 : int32_t ThreadPool::SetAffinityBySelf(const size_t threadIndex, const uint32_t deviceId)
357 : {
358 22 : if (hasAicpu_ && (AddPidToTask(threadIndex) != AICPU_SCHEDULE_OK)) {
359 2 : aicpusd_err("AddPidToTask failed");
360 2 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
361 : }
362 : cpu_set_t mask;
363 20 : CPU_ZERO(&mask);
364 20 : const uint32_t aicpuLogIndex = hasAicpu_?
365 16 : static_cast<uint32_t>(threadIndex) % AicpuDrvManager::GetInstance().GetAicpuNumPerDevice(): 0U;
366 :
367 20 : uint32_t physIndex = 0;
368 20 : uint32_t devNum = 0U;
369 20 : if ((FeatureCtrl::IsVfModeCheckedByDeviceId(deviceId)) && (&halGetVdevNum != nullptr)) {
370 3 : const int32_t result = halGetVdevNum(&devNum);
371 3 : if (result != 0) {
372 1 : aicpusd_err("halGetVdevNum, failed result[%d]", result);
373 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
374 : }
375 : }
376 19 : if (!hasAicpu_) {
377 4 : physIndex = GetNoAicpuCcpuPhysIndex(threadIndex, deviceId);
378 15 : } else if (devNum > 0U) {
379 1 : physIndex = AicpuDrvManager::GetInstance().GetAicpuPhysIndexInVfMode(aicpuLogIndex, deviceId);
380 : } else {
381 14 : physIndex = AicpuDrvManager::GetInstance().GetAicpuPhysIndex(aicpuLogIndex, deviceId);
382 : }
383 19 : aicpusd_info("[hw]SetAffinityBySelf, physIndex[%u], devNum[%u]", physIndex, devNum);
384 19 : if (physIndex == INVALID_AICPU_ID) {
385 0 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_RUNNING;
386 0 : return static_cast<int32_t>(AICPU_SCHEDULE_OK);
387 : }
388 : // cannot overflow, aicpu num < 65535, max [64=4*16]
389 19 : CPU_SET(static_cast<int32_t>(physIndex), &mask);
390 19 : const int32_t ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &mask);
391 19 : if (ret != 0) {
392 1 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_EXIT;
393 1 : aicpusd_err("set affinity failed ret[%d], aicpu logical index[%zu], aicpu physical index[%u], "
394 : "device id[%u]", ret, threadIndex, physIndex, deviceId);
395 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
396 : }
397 18 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_RUNNING;
398 18 : aicpusd_info("set affinity success, aicpu logical index[%zu], aicpu physical index[%u], device id[%u]",
399 : threadIndex, physIndex, deviceId);
400 18 : return AICPU_SCHEDULE_OK;
401 : }
402 :
403 7 : int32_t ThreadPool::SetAffinityByPm(const size_t threadIndex, const uint32_t deviceId)
404 : {
405 7 : const uint32_t aicpuLogIndex = hasAicpu_?
406 3 : static_cast<uint32_t>(threadIndex) % AicpuDrvManager::GetInstance().GetAicpuNumPerDevice(): 0U;
407 : uint32_t physIndex;
408 7 : if (FeatureCtrl::IsVfModeCheckedByDeviceId(deviceId)) {
409 2 : physIndex = AicpuDrvManager::GetInstance().GetAicpuPhysIndex(aicpuLogIndex,
410 : (deviceId - VDEVICE_MIN_CPU_NUM));
411 : } else {
412 10 : physIndex = hasAicpu_?
413 2 : AicpuDrvManager::GetInstance().GetAicpuPhysIndex(aicpuLogIndex, deviceId):
414 3 : GetNoAicpuCcpuPhysIndex(threadIndex, deviceId);
415 : }
416 7 : const pid_t tid = static_cast<pid_t>(GetTid());
417 :
418 7 : std::vector<uint32_t> coreAffinity;
419 7 : coreAffinity.push_back(physIndex);
420 7 : aicpusd_info("begin to ProcMgrBindThread, tid:%d, physIndex:%u, hasAicpu_:%d", tid, physIndex, hasAicpu_);
421 7 : auto ret = ProcMgrBindThread(tid, coreAffinity);
422 7 : aicpusd_info("end to ProcMgrBindThread, ret:%d", ret);
423 7 : uint32_t tryTimes = 0;
424 9 : while ((ret != 0U) && (tryTimes <= 1)) {
425 2 : aicpusd_warn("set affinity failed ret[%d], will try again", ret);
426 2 : (void)usleep(SLEEP_USECS);
427 2 : ret = ProcMgrBindThread(tid, coreAffinity);
428 2 : tryTimes++;
429 : }
430 7 : if (ret != 0U) {
431 1 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_EXIT;
432 1 : aicpusd_err("set affinity failed ret[%d], aicpu logical index[%zu], "
433 : "aicpu physical index[%u],tid[%u], device id[%u]",
434 : ret, threadIndex, physIndex, tid, AicpuDrvManager::GetInstance().GetDeviceId());
435 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
436 : }
437 6 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_RUNNING;
438 6 : aicpusd_info("set affinity success, aicpu logical index[%zu], aicpu physical index[%u], device id[%u]",
439 : threadIndex, physIndex, AicpuDrvManager::GetInstance().GetDeviceId());
440 6 : return AICPU_SCHEDULE_OK;
441 7 : }
442 :
443 28 : int32_t ThreadPool::SetAffinity(const size_t threadIndex, const uint32_t deviceId)
444 : {
445 28 : int32_t res = static_cast<int32_t>(AICPU_SCHEDULE_OK);
446 56 : if (AicpuUtil::IsEnvValEqual(ENV_NAME_PROCMGR_AICPU_CPUSET, "1")) {
447 7 : res = SetAffinityByPm(threadIndex, deviceId);
448 7 : aicpusd_run_info("aicpu bind tid by pm, index[%zu], deviceId[%u], res[%d].", threadIndex, deviceId, res);
449 : } else {
450 21 : res = SetAffinityBySelf(threadIndex, deviceId);
451 21 : aicpusd_info("aicpu bind tid by self, index[%zu], deviceId[%u], res[%d].", threadIndex, deviceId, res);
452 : }
453 28 : return res;
454 : }
455 :
456 1 : void ThreadPool::SetThreadStatus(const size_t threadIndex, const ThreadStatus threadStat)
457 : {
458 1 : threadStatusList_[threadIndex] = threadStat;
459 1 : }
460 :
461 12 : void ThreadPool::PostSem(const size_t threadIndex)
462 : {
463 12 : (void)sem_post(&(sems_[threadIndex]));
464 12 : }
465 :
466 15 : void ThreadPool::SetThreadIdRelation(const size_t threadIndex, const pid_t threadId)
467 : {
468 15 : threadIdLists_[threadIndex] = threadId;
469 15 : aicpusd_info("set thread index:%zu and tid:%d relation to List", threadIndex,
470 : static_cast<int32_t>(threadIdLists_[threadIndex]));
471 15 : }
472 :
473 9 : void ThreadPool::SetThreadSchedModeByTsd()
474 : {
475 9 : const size_t relationSize = threadIdLists_.size();
476 9 : if (relationSize > MAX_THREAD_ID_CNT) {
477 1 : aicpusd_err("current list to long size:%zu", relationSize);
478 3 : return;
479 : }
480 8 : SubProcScheduleModeInfo curInfo = {};
481 8 : curInfo.totalNum = static_cast<uint32_t>(relationSize);
482 8 : (void)std::copy(threadIdLists_.begin(), threadIdLists_.end(), curInfo.threadIdList);
483 8 : const std::vector<uint32_t> deviceVec = AicpuDrvManager::GetInstance().GetDeviceList();
484 8 : if (deviceVec.empty()) {
485 1 : aicpusd_err("the device vector is empty");
486 1 : return;
487 : }
488 7 : if (SetSubProcScheduleMode(deviceVec[0],
489 : static_cast<uint32_t>(TsdWaitType::TSD_COMPUTE),
490 7 : static_cast<uint32_t>(AicpuDrvManager::GetInstance().GetHostPid()),
491 14 : AicpuDrvManager::GetInstance().GetVfId(), &curInfo) != 0) {
492 1 : aicpusd_err("send msg to tsd failed");
493 1 : return;
494 : }
495 8 : }
496 : } // namespace AicpuSchedule
|
