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 :
18 : #include "aicpusd_status.h"
19 : #include "aicpusd_util.h"
20 : #include "aicpusd_drv_manager.h"
21 : #include "aicpusd_monitor.h"
22 : #include "aicpusd_event_manager.h"
23 : #include "aicpusd_context.h"
24 : #include "aicpu_context.h"
25 : #include "aicpusd_proc_mgr_sys_operator_agent.h"
26 : #include "aicpusd_hal_interface_ref.h"
27 : #include "aicpusd_so_manager.h"
28 : #include "aicpu_pulse.h"
29 : #include "aicpusd_feature_ctrl.h"
30 :
31 : namespace {
32 : constexpr uint32_t CP_EVENT_MASK =
33 : static_cast<uint32_t>(static_cast<uint32_t>(1U) << static_cast<uint32_t>(EVENT_TS_CTRL_MSG));
34 : }
35 :
36 : namespace AicpuSchedule {
37 22 : ThreadPool &ThreadPool::Instance()
38 : {
39 22 : static ThreadPool threadPoolInstance;
40 22 : return threadPoolInstance;
41 : }
42 :
43 8 : ThreadPool::ThreadPool() : semInitedNum_(0UL)
44 : {
45 8 : aicpusd_run_info("ThreadPool");
46 8 : }
47 :
48 8 : ThreadPool::~ThreadPool()
49 : {
50 8 : Clear();
51 8 : }
52 :
53 14 : void ThreadPool::Clear()
54 : {
55 14 : AicpuSchedule::AicpuEventManager::GetInstance().SetRunningFlag(false);
56 14 : WaitForStop();
57 20 : for (auto &sem : sems_) {
58 6 : (void)sem_destroy(&sem);
59 : }
60 14 : semInitedNum_ = 0UL;
61 14 : threadStatusList_.clear();
62 14 : threadIdLists_.clear();
63 14 : }
64 :
65 6 : int32_t ThreadPool::CreateWorker(const AicpuSchedMode schedMode)
66 : {
67 6 : Clear();
68 6 : AicpuSchedule::AicpuEventManager::GetInstance().SetRunningFlag(true);
69 6 : schedMode_ = schedMode;
70 6 : const size_t aicpuNum = AicpuDrvManager::GetInstance().GetAicpuNum();
71 6 : const std::vector<uint32_t> deviceVec = AicpuDrvManager::GetInstance().GetDeviceList();
72 6 : if (aicpuNum == 0UL) {
73 1 : aicpusd_run_info("aicpu total num[0], need not create aicpu worker");
74 : } else {
75 : try {
76 5 : sems_ = std::move(std::vector<sem_t>(static_cast<size_t>(aicpuNum)));
77 0 : } catch (std::exception &threadException) {
78 0 : aicpusd_err("create sems failed, %s", threadException.what());
79 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
80 0 : }
81 9 : for (size_t i = 0UL; i < aicpuNum; ++i) {
82 5 : const int32_t semInitRet = sem_init(&(sems_[i]), 0, 0U);
83 5 : if (semInitRet == -1) {
84 1 : aicpusd_err("sem[%zu] init failed, %s", i, strerror(errno));
85 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
86 : }
87 4 : semInitedNum_ = i + 1UL;
88 : }
89 : try {
90 8 : threadStatusList_ = std::move(std::vector<ThreadStatus>(static_cast<size_t>(aicpuNum),
91 12 : ThreadStatus::THREAD_INIT));
92 0 : } catch (std::exception &threadException) {
93 0 : aicpusd_err("create ThreadStatus failed, %s", threadException.what());
94 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
95 0 : }
96 4 : int32_t ret = AICPU_SCHEDULE_OK;
97 4 : const sighandler_t oldHandler = signal(SIGCHLD, SIG_DFL);
98 4 : aicpusd_info("set SIGCHLD to %d, old sighandler[%d]", SIG_DFL, oldHandler);
99 4 : const size_t aicpuNumPerDev = AicpuDrvManager::GetInstance().GetAicpuNumPerDevice();
100 7 : for (size_t i = 0UL; i < aicpuNum; ++i) {
101 : // aicpuNumPerDev is not 0
102 4 : const size_t deviceVecInx = i / aicpuNumPerDev;
103 4 : ret = CreateOneWorker(i, deviceVec[deviceVecInx]);
104 4 : if (ret != AICPU_SCHEDULE_OK) {
105 1 : (void)signal(SIGCHLD, oldHandler);
106 1 : return ret;
107 : }
108 : }
109 :
110 3 : for (size_t i = 0UL; i < aicpuNum; i++) {
111 3 : const int32_t semWaitRet = sem_wait(&(sems_[i]));
112 3 : if (semWaitRet == -1) {
113 1 : (void)signal(SIGCHLD, oldHandler);
114 1 : aicpusd_err("sem[%zu] wait failed, %s", i, strerror(errno));
115 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
116 : }
117 2 : if (threadStatusList_[i] != ThreadStatus::THREAD_RUNNING) {
118 2 : (void)signal(SIGCHLD, oldHandler);
119 2 : aicpusd_err("create thread[%zu] failed", i);
120 2 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
121 : }
122 : }
123 0 : aicpusd_info("set SIGCHLD to old sighandler[%d]", oldHandler);
124 0 : (void)signal(SIGCHLD, oldHandler);
125 : }
126 :
127 1 : return AICPU_SCHEDULE_OK;
128 6 : }
129 :
130 2 : int32_t ThreadPool::CreateOneWorker(const size_t threadIndex,
131 : const uint32_t deviceId)
132 : {
133 : try {
134 2 : aicpusd_info("CreateOneWorker device[%u]:thread[%zu] started.", deviceId, threadIndex);
135 2 : std::thread th(&ThreadPool::Work, threadIndex, deviceId, schedMode_);
136 2 : pthread_setname_np(th.native_handle(), "aicpu_dump");
137 2 : workers_.emplace_back(std::move(th));
138 2 : } catch (std::exception &threadException) {
139 0 : aicpusd_err("create aicpu worker[%zu] failed, %s", threadIndex, threadException.what());
140 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
141 0 : }
142 :
143 2 : return AICPU_SCHEDULE_OK;
144 : }
145 :
146 14 : void ThreadPool::WaitForStop()
147 : {
148 14 : aicpusd_run_info("WaitForStop begin.");
149 18 : for (auto &worker : workers_) {
150 4 : if (worker.joinable()) {
151 2 : worker.join();
152 : }
153 : }
154 14 : aicpusd_run_info("WaitForStop end.");
155 14 : }
156 :
157 8 : void ThreadPool::Work(const size_t threadIndex, const uint32_t deviceId, const AicpuSchedMode schedMode)
158 : {
159 : (void)schedMode;
160 8 : aicpusd_info("Aicpu device[%u]:thread[%zu] started.", deviceId, threadIndex);
161 : aicpu::aicpuContext_t context;
162 8 : context.tsId = 0U;
163 8 : context.hostPid = AicpuDrvManager::GetInstance().GetHostPid();
164 8 : context.vfId = AicpuDrvManager::GetInstance().GetVfId();
165 8 : context.deviceId = deviceId;
166 8 : if (aicpu::aicpuSetContext(&context) != aicpu::AICPU_ERROR_NONE) {
167 1 : aicpusd_err("Set aicpu context failed, deviceId[%u], thread[%zu].", deviceId, threadIndex);
168 1 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
169 5 : return;
170 : }
171 7 : aicpusd_info("halEschedSubscribeEvent success, deviceId[%u], groupId[%u], threadIndex[%zu] eventBitmap[%llu].",
172 : deviceId, CP_DEFAULT_GROUP_ID, threadIndex, CP_EVENT_MASK);
173 7 : DeployContext deployCtx = DeployContext::DEVICE;
174 7 : const StatusCode ctxRet = GetAicpuDeployContext(deployCtx);
175 7 : if (ctxRet != AICPU_SCHEDULE_OK) {
176 1 : aicpusd_err("Get current deploy ctx failed.");
177 1 : return;
178 : }
179 :
180 6 : if (deployCtx == DeployContext::DEVICE) {
181 6 : if (AicpuSchedule::ThreadPool::Instance().SetAffinity(threadIndex, deviceId)
182 6 : != AICPU_SCHEDULE_OK) {
183 1 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
184 1 : return;
185 : }
186 : } else {
187 0 : AicpuSchedule::ThreadPool::Instance().SetThreadStatus(threadIndex, ThreadStatus::THREAD_RUNNING);
188 : }
189 5 : const int32_t ret = halEschedSubscribeEvent(deviceId, CP_DEFAULT_GROUP_ID, static_cast<uint32_t>(threadIndex),
190 5 : CP_EVENT_MASK);
191 5 : if (ret != DRV_ERROR_NONE) {
192 2 : aicpusd_err("halEschedSubscribeEvent failed, deviceId[%u], groupId[%u], threadIndex[%zu] "
193 : "eventBitmap[%llu].", deviceId, CP_DEFAULT_GROUP_ID, threadIndex, CP_EVENT_MASK);
194 2 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
195 2 : return;
196 : }
197 :
198 3 : AicpuSchedule::ThreadPool::Instance().PostSem(threadIndex);
199 :
200 3 : AicpuSchedule::AicpuEventManager::GetInstance().LoopProcess(static_cast<uint32_t>(threadIndex));
201 3 : aicpusd_info("Aicpu device[%u]:thread[%u] stopped.", deviceId, threadIndex);
202 : }
203 :
204 7 : int32_t ThreadPool::WriteTidForAffinity(const size_t threadIndex)
205 : {
206 7 : if (threadIndex >= threadStatusList_.size()) {
207 1 : aicpusd_err("threadIndex[%zu], out of rank[0, %zu]",
208 : threadIndex, threadStatusList_.size());
209 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
210 : }
211 :
212 12 : std::string command = "cd /var/ && sudo ./add_aicpu_tid_to_tasks.sh";
213 6 : std::string pathStr = "/var/add_aicpu_tid_to_tasks.sh";
214 6 : if (access(pathStr.c_str(), F_OK) != 0) {
215 5 : aicpusd_info("Not find add_aicpu_tid_to_tasks.sh.");
216 5 : return AICPU_SCHEDULE_OK;
217 : }
218 1 : command = command + " " + std::to_string(GetTid());
219 :
220 : // 使用system命令会对父进程进行拷贝，浪费了系统资源。在esl等环境中还会存在由于资源较少无法fork导致system卡住的问题.
221 : // 使用vfork替换system命令，由于与父进程共享资源，因此可解决资源浪费/卡住的问题.
222 1 : const int32_t ret = AicpuUtil::ExecuteCmd(command);
223 1 : if (ret != 0) {
224 1 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_EXIT;
225 1 : aicpusd_err("write tid[%llu] to /sys/fs/cgroup/cpuset/AICPU/tasks failed, ret[%d], strerror[%s]",
226 : GetTid(), ret, strerror(errno));
227 1 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
228 : }
229 :
230 0 : return AICPU_SCHEDULE_OK;
231 6 : }
232 :
233 6 : int32_t ThreadPool::AddPidToTask(const size_t threadIndex)
234 : {
235 6 : if (FeatureCtrl::IsBindPidByHal()) {
236 2 : if (&halBindCgroup != nullptr) {
237 2 : aicpusd_info("Bind pid by hal index:%zu.", threadIndex);
238 2 : const drvError_t drvRet = halBindCgroup(BIND_AICPU_CGROUP);
239 2 : if (drvRet != DRV_ERROR_NONE) {
240 1 : aicpusd_err("halBindCgroup failed, ret[%d]", drvRet);
241 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
242 : }
243 1 : aicpusd_info("halBindCgroup success");
244 : }
245 : } else {
246 4 : aicpusd_run_info("AddPidToTask by WriteTidForAffinity");
247 4 : auto ret = WriteTidForAffinity(threadIndex);
248 4 : if (ret != static_cast<int32_t>(AICPU_SCHEDULE_OK)) {
249 0 : aicpusd_err("WriteTidForAffinity failed, ret[%d]", ret);
250 0 : return static_cast<int32_t>(AICPU_SCHEDULE_ERROR_INIT_FAILED);
251 : }
252 4 : aicpusd_info("WriteTidForAffinity success");
253 : }
254 5 : return AICPU_SCHEDULE_OK;
255 : }
256 :
257 4 : int32_t ThreadPool::SetAffinityBySelf(const size_t threadIndex, const uint32_t deviceId)
258 : {
259 4 : if (AddPidToTask(threadIndex) != AICPU_SCHEDULE_OK) {
260 0 : aicpusd_err("AddPidToTask failed");
261 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
262 : }
263 : cpu_set_t mask;
264 4 : CPU_ZERO(&mask);
265 4 : if (AicpuDrvManager::GetInstance().GetAicpuNumPerDevice() == 0) {
266 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
267 : }
268 4 : const uint32_t aicpuLogIndex = static_cast<uint32_t>(threadIndex) %
269 4 : AicpuDrvManager::GetInstance().GetAicpuNumPerDevice();
270 :
271 4 : uint32_t physIndex = 0;
272 4 : uint32_t devNum = 0U;
273 4 : if ((FeatureCtrl::IsVfModeCheckedByDeviceId(deviceId)) && (&halGetVdevNum != nullptr)) {
274 0 : int32_t result = halGetVdevNum(&devNum);
275 0 : if (result != 0) {
276 0 : aicpusd_err("halGetVdevNum, failed result[%d]", result);
277 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
278 : }
279 : }
280 4 : if (devNum > 0U) {
281 0 : physIndex = AicpuDrvManager::GetInstance().GetAicpuPhysIndexInVfMode(aicpuLogIndex, deviceId);
282 : } else {
283 4 : physIndex = AicpuDrvManager::GetInstance().GetAicpuPhysIndex(aicpuLogIndex, deviceId);
284 : }
285 4 : aicpusd_run_info("[hw]SetAffinityBySelf, physIndex[%u], devNum[%u]", physIndex, devNum);
286 4 : if (physIndex == INVALID_AICPU_ID) {
287 0 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_RUNNING;
288 0 : return static_cast<int32_t>(AICPU_SCHEDULE_OK);
289 : }
290 : // cannot overflow, aicpu num < 65535, max [64=4*16]
291 4 : CPU_SET(static_cast<int32_t>(physIndex), &mask);
292 4 : const int32_t ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &mask);
293 4 : if (ret != 0) {
294 0 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_EXIT;
295 0 : aicpusd_err("set affinity failed ret[%d], aicpu logical index[%zu], aicpu physical index[%u], "
296 : "device id[%u]", ret, threadIndex, physIndex, deviceId);
297 0 : return AICPU_SCHEDULE_ERROR_INIT_FAILED;
298 : }
299 4 : threadStatusList_[threadIndex] = ThreadStatus::THREAD_RUNNING;
300 4 : aicpusd_info("set affinity success, aicpu logical index[%zu], aicpu physical index[%u], device id[%u]",
301 : threadIndex, physIndex, deviceId);
302 4 : return AICPU_SCHEDULE_OK;
303 : }
304 :
305 4 : int32_t ThreadPool::SetAffinity(const size_t threadIndex, const uint32_t deviceId)
306 : {
307 4 : int32_t res = static_cast<int32_t>(AICPU_SCHEDULE_OK);
308 8 : if (AicpuUtil::IsEnvValEqual(ENV_NAME_PROCMGR_AICPU_CPUSET, "1")) {
309 0 : aicpusd_err("aicpu bind tid by pm, index[%zu], deviceId[%u].", threadIndex, deviceId);
310 : } else {
311 4 : res = SetAffinityBySelf(threadIndex, deviceId);
312 4 : aicpusd_run_info("aicpu bind tid by self, index[%zu], deviceId[%u], res[%d].", threadIndex, deviceId, res);
313 : }
314 4 : return res;
315 : }
316 :
317 0 : void ThreadPool::SetThreadStatus(const size_t threadIndex, const ThreadStatus threadStat)
318 : {
319 0 : threadStatusList_[threadIndex] = threadStat;
320 0 : }
321 :
322 3 : void ThreadPool::PostSem(const size_t threadIndex)
323 : {
324 3 : (void)sem_post(&(sems_[threadIndex]));
325 3 : }
326 : } // namespace AicpuSchedule
|
