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