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