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 : #include "aicpusd_threads_process.h"
11 :
12 : #include "ascend_hal.h"
13 : #include "profiling_adp.h"
14 : #include "aicpusd_status.h"
15 : #include "aicpusd_event_manager.h"
16 : #include "aicpu_context.h"
17 : #include "aicpusd_common.h"
18 : #include "aicpusd_drv_manager.h"
19 : #include "aicpusd_worker.h"
20 : #include "aicpusd_monitor.h"
21 : #include "aicpusd_hal_interface_ref.h"
22 : #include "aicpu_prof.h"
23 : #include "aicpu_cust_sd_dump_process.h"
24 : #include "aicpusd_util.h"
25 :
26 : namespace AicpuSchedule {
27 18 : ComputeProcess::ComputeProcess()
28 18 : : deviceId_(0U),
29 18 : hostPid_(-1),
30 18 : aicpuNum_(0U),
31 18 : profilingMode_(PROFILING_CLOSE),
32 18 : aicpuPid_(-1),
33 18 : vfId_(0U),
34 18 : runMode_(aicpu::AicpuRunMode::THREAD_MODE)
35 18 : {}
36 :
37 25 : ComputeProcess& ComputeProcess::GetInstance()
38 : {
39 25 : static ComputeProcess instance;
40 25 : return instance;
41 : }
42 :
43 8 : void ComputeProcess::UpdateProfilingSetting(uint32_t flag)
44 : {
45 8 : ProfilingMode profilingMode = PROFILING_CLOSE;
46 8 : bool kernelFlag = false;
47 8 : AicpuUtil::GetProfilingInfo(flag, profilingMode, kernelFlag);
48 8 : profilingMode_ = profilingMode;
49 8 : if (kernelFlag) {
50 1 : aicpu::UpdateMode(profilingMode == PROFILING_OPEN);
51 : }
52 8 : aicpusd_info(
53 : "Update aicpu profiling mode success, flag[%u], profilingMode[%u], kernelFlag[%d],", flag, profilingMode,
54 : kernelFlag);
55 8 : }
56 :
57 12 : int32_t ComputeProcess::Start(
58 : const uint32_t deviceId, const pid_t hostPid, const uint32_t profilingMode, const pid_t aicpuPid,
59 : const uint32_t vfId, const aicpu::AicpuRunMode runMode)
60 : {
61 12 : aicpusd_info(
62 : "AicpuCustSd start, deviceId[%u] hostpid[%d] profilingMode[%u] aicpuPid[%d] runMode[%d] vfId[%u].", deviceId,
63 : hostPid, profilingMode, aicpuPid, static_cast<int32_t>(runMode), vfId);
64 :
65 12 : const AicpuSchedule::AicpuDrvManager& drvMgr = AicpuSchedule::AicpuDrvManager::GetInstance();
66 12 : aicpuNum_ = drvMgr.GetAicpuNum();
67 :
68 12 : deviceId_ = deviceId;
69 12 : hostPid_ = hostPid;
70 12 : runMode_ = runMode;
71 12 : aicpuPid_ = aicpuPid;
72 12 : vfId_ = vfId;
73 :
74 12 : if ((aicpuNum_ > 0U) && (aicpu::InitTaskMonitorContext(aicpuNum_) != aicpu::AICPU_ERROR_NONE)) {
75 1 : aicpusd_err("Init task monitor context failed");
76 1 : return static_cast<int32_t>(ComputProcessRetCode::CP_RET_COMMON_ERROR);
77 : }
78 11 : if (runMode_ == aicpu::AicpuRunMode::PROCESS_PCIE_MODE) {
79 4 : if (&halMemBindSibling == nullptr) {
80 0 : aicpusd_err("Interface halMemBindSibling is not supported in current device");
81 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
82 : }
83 4 : drvError_t drvRet = DRV_ERROR_NONE;
84 4 : if (AicpuSchedule::AicpuDrvManager::GetInstance().GetSafeVerifyFlag()) {
85 3 : drvRet = halMemBindSibling(hostPid, aicpuPid, vfId_, deviceId, SVM_MEM_BIND_SVM_GRP);
86 : } else {
87 : // new
88 1 : aicpusd_info("open prof so and bind sp group without alloc permission.");
89 1 : aicpu::LoadProfilingLib();
90 1 : drvRet = halMemBindSibling(
91 1 : hostPid, aicpuPid, vfId_, deviceId, SVM_MEM_BIND_SVM_GRP | SVM_MEM_BIND_SP_GRP_NO_ALLOC);
92 : }
93 :
94 4 : if (drvRet != DRV_ERROR_NONE) {
95 3 : aicpusd_err(
96 : "Failed to halMemBindSibling, hostpid[%d], aicpusd pid[%d], deviceId[%u], vfId[%u], ret[%d].", hostPid,
97 : aicpuPid, deviceId, vfId, drvRet);
98 3 : return static_cast<int32_t>(ComputProcessRetCode::CP_RET_COMMON_ERROR);
99 : }
100 1 : aicpusd_info(
101 : "Bind Sibling pid success, hostpid[%d] aicpusd pid[%d] deviceId[%u] vfId[%u].", hostPid, aicpuPid, deviceId,
102 : vfId);
103 : }
104 8 : aicpu::InitProfilingDataInfo(deviceId_, hostPid, CHANNEL_CUS_AICPU);
105 8 : UpdateProfilingSetting(profilingMode);
106 8 : if (profilingMode != 0U) {
107 8 : aicpu::LoadProfilingLib();
108 8 : aicpu::SetProfilingFlagForKFC(profilingMode);
109 8 : aicpu::UpdateMode((profilingMode & 1) == PROFILING_OPEN);
110 : }
111 8 : (void)aicpu::SetAicpuRunMode(runMode_);
112 8 : aicpu::SetCustAicpuSdFlag(true);
113 8 : const uint32_t ret = RegisterScheduleTask();
114 8 : if (ret != AICPU_SCHEDULE_OK) {
115 1 : aicpusd_err("Register aicpu split and random kernel scheduler failed.");
116 1 : return static_cast<int32_t>(ret);
117 : }
118 :
119 7 : const int32_t aicpuStartRet = AicpuSchedule::ThreadPool::Instance().CreateWorker();
120 7 : if (aicpuStartRet != AICPU_SCHEDULE_OK) {
121 1 : aicpusd_err("Drv create aicpu work tasks failed, ret[%d].", aicpuStartRet);
122 1 : return static_cast<int32_t>(ComputProcessRetCode::CP_RET_COMMON_ERROR);
123 : }
124 6 : AicpuCustDumpProcess::GetInstance().InitDumpProcess(
125 6 : deviceId, static_cast<uint32_t>(AicpuSchedule::ThreadPool::GetWorkerNum()));
126 6 : aicpusd_info(
127 : "Aicpu custom scheduler start succeed, deviceId[%u], hostpid[%d], profilingMode[%u], runMode[%d].", deviceId,
128 : hostPid, profilingMode, runMode_);
129 6 : return static_cast<int32_t>(ComputProcessRetCode::CP_RET_SUCCESS);
130 : }
131 :
132 24 : uint32_t ComputeProcess::RegisterScheduleTask()
133 : {
134 3 : const auto randomKernelScheduler = [this](const aicpu::Closure& task) { return SubmitRandomKernelTask(task); };
135 : const auto splitKernelScheduler =
136 10 : [this](const uint32_t parallelId, const int64_t shardNum, const std::queue<aicpu::Closure>& taskQueue) {
137 10 : const AICPUSharderTaskInfo taskInfo = {.parallelId = parallelId, .shardNum = shardNum};
138 20 : return SubmitSplitKernelTask(taskInfo, taskQueue);
139 24 : };
140 3 : const auto splitKernelGetProcesser = [this]() { return GetAndDoSplitKernelTask(); };
141 :
142 48 : aicpu::SharderNonBlock::GetInstance().Register(
143 48 : aicpuNum_, randomKernelScheduler, splitKernelScheduler, splitKernelGetProcesser);
144 :
145 24 : return AICPU_SCHEDULE_OK;
146 : }
147 :
148 3 : uint32_t ComputeProcess::SubmitRandomKernelTask(const aicpu::Closure& task)
149 : {
150 3 : if (!randomKernelTask_.Enqueue(task)) {
151 1 : aicpusd_err("Add random kernel task failed.");
152 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
153 : }
154 :
155 2 : AICPUSubEventInfo aicpuEventInfo = {};
156 2 : event_summary eventInfoSummary = {};
157 2 : eventInfoSummary.pid = getpid();
158 2 : eventInfoSummary.event_id = EVENT_RANDOM_KERNEL;
159 2 : eventInfoSummary.msg = PtrToPtr<AICPUSubEventInfo, char_t>(&aicpuEventInfo);
160 2 : eventInfoSummary.msg_len = static_cast<uint32_t>(sizeof(AICPUSubEventInfo));
161 :
162 2 : const int32_t drvRet = halEschedSubmitEvent(deviceId_, &eventInfoSummary);
163 2 : if (drvRet != DRV_ERROR_NONE) {
164 1 : aicpusd_err("Submit random kernel event failed. ret=%d", drvRet);
165 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
166 : }
167 :
168 1 : return AICPU_SCHEDULE_OK;
169 : }
170 :
171 10 : uint32_t ComputeProcess::SubmitSplitKernelTask(
172 : const AICPUSharderTaskInfo& taskInfo, const std::queue<aicpu::Closure>& taskQueue)
173 : {
174 10 : if (!splitKernelTask_.BatchAddTask(taskInfo, taskQueue)) {
175 1 : aicpusd_err("Add split kernel task to map failed, parallelId=%u", taskInfo.parallelId);
176 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
177 : }
178 :
179 9 : uint32_t ret = AICPU_SCHEDULE_OK;
180 9 : if (FeatureCtrl::ShouldSubmitTaskOneByOne()) {
181 2 : ret = SubmitBatchSplitKernelEventOneByOne(taskInfo);
182 : } else {
183 7 : ret = SubmitBatchSplitKernelEventDc(taskInfo);
184 : }
185 9 : if (ret != AICPU_SCHEDULE_OK) {
186 2 : aicpusd_err(
187 : "Submit batch split kernel event failed. parallelId=%u, submitNum=%ld", taskInfo.parallelId,
188 : taskInfo.shardNum);
189 2 : return ret;
190 : }
191 :
192 7 : aicpusd_info(
193 : "Submit split kernel event success. parallelId=%u, submitNum=%ld", taskInfo.parallelId, taskInfo.shardNum);
194 :
195 7 : return AICPU_SCHEDULE_OK;
196 : }
197 :
198 2 : uint32_t ComputeProcess::SubmitBatchSplitKernelEventOneByOne(const AICPUSharderTaskInfo& taskInfo) const
199 : {
200 2 : const uint32_t submitNum = static_cast<uint32_t>(taskInfo.shardNum);
201 3 : for (uint32_t i = 0U; i < submitNum; ++i) {
202 2 : const uint32_t ret = SubmitOneSplitKernelEvent(taskInfo);
203 2 : if (ret != AICPU_SCHEDULE_OK) {
204 1 : aicpusd_err("Submit single split kernel event failed. parallelId=%u, i=%u", taskInfo.parallelId, i);
205 1 : return ret;
206 : }
207 : }
208 :
209 1 : return AICPU_SCHEDULE_OK;
210 : }
211 :
212 7 : uint32_t ComputeProcess::SubmitBatchSplitKernelEventDc(const AICPUSharderTaskInfo& taskInfo)
213 : {
214 7 : AICPUSubEventInfo aicpuEventInfo = {};
215 7 : aicpuEventInfo.para.sharderTaskInfo = taskInfo;
216 7 : event_summary eventInfoSummary = {};
217 7 : eventInfoSummary.pid = getpid();
218 7 : eventInfoSummary.event_id = EVENT_SPLIT_KERNEL;
219 7 : eventInfoSummary.msg = PtrToPtr<AICPUSubEventInfo, char_t>(&aicpuEventInfo);
220 7 : eventInfoSummary.msg_len = static_cast<uint32_t>(sizeof(AICPUSubEventInfo));
221 7 : uint32_t submitSuccessNum = 0U;
222 7 : const uint32_t submitNum = static_cast<uint32_t>(taskInfo.shardNum);
223 : const int32_t drvRet =
224 7 : halEschedSubmitEventBatch(deviceId_, SHARED_EVENT_ENTRY, &eventInfoSummary, submitNum, &submitSuccessNum);
225 7 : if ((drvRet == DRV_ERROR_NONE) && (submitSuccessNum == submitNum)) {
226 4 : aicpusd_info(
227 : "Batch submit split kernel event success, parallelId=%u, submitNum=%u", taskInfo.parallelId, submitNum);
228 4 : return AICPU_SCHEDULE_OK;
229 : }
230 :
231 : /*
232 : * The queue depth of event schedule is only dozens. If too many split kernel event are submited,
233 : * the queue will be full. Therefore, the main thread needs to process the task sending failure.
234 : */
235 3 : aicpusd_warn(
236 : "Batch submit some of split kernel event success, ret=%d, parallelId=%u, submitNum=%u, "
237 : "submitSuccessNum=%u",
238 : drvRet, taskInfo.parallelId, submitNum, submitSuccessNum);
239 :
240 3 : const uint32_t remainNum = (drvRet == DRV_ERROR_NONE) ? submitNum - submitSuccessNum : submitNum;
241 5 : for (uint32_t i = 0U; i < remainNum; ++i) {
242 3 : if (!DoSplitKernelTask(taskInfo)) {
243 1 : aicpusd_err(
244 : "Run single task failed after batch submit fail, parallelId=%u, submitNum=%u, "
245 : "remainNum=%u, i=%u",
246 : taskInfo.parallelId, submitNum, remainNum, i);
247 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
248 : }
249 : }
250 :
251 2 : return AICPU_SCHEDULE_OK;
252 : }
253 :
254 2 : uint32_t ComputeProcess::SubmitOneSplitKernelEvent(const AICPUSharderTaskInfo& taskInfo) const
255 : {
256 2 : AICPUSubEventInfo aicpuEventInfo = {};
257 2 : aicpuEventInfo.para.sharderTaskInfo = taskInfo;
258 2 : event_summary eventInfoSummary = {};
259 2 : eventInfoSummary.pid = getpid();
260 2 : eventInfoSummary.event_id = EVENT_SPLIT_KERNEL;
261 2 : eventInfoSummary.msg = PtrToPtr<AICPUSubEventInfo, char_t>(&aicpuEventInfo);
262 2 : eventInfoSummary.msg_len = static_cast<uint32_t>(sizeof(AICPUSubEventInfo));
263 :
264 2 : const int32_t drvRet = halEschedSubmitEvent(deviceId_, &eventInfoSummary);
265 2 : if (drvRet != DRV_ERROR_NONE) {
266 1 : aicpusd_err("Submit split kernel event failed. ret=%d, parallelId=%u", drvRet, taskInfo.parallelId);
267 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
268 : }
269 :
270 1 : return AICPU_SCHEDULE_OK;
271 : }
272 :
273 3 : bool ComputeProcess::GetAndDoSplitKernelTask()
274 : {
275 3 : event_info eventInfo = {};
276 3 : const uint32_t threadIndex = aicpu::GetAicpuThreadIndex();
277 : const int32_t retVal =
278 3 : halEschedGetEvent(deviceId_, AicpuSchedule::DEFAULT_GROUP_ID, threadIndex, EVENT_SPLIT_KERNEL, &eventInfo);
279 3 : if (retVal == DRV_ERROR_NO_EVENT) {
280 1 : return true;
281 : }
282 2 : if (retVal != DRV_ERROR_NONE) {
283 1 : aicpusd_err("Cannot get event, threadIndex=%u, ret=%d", threadIndex, retVal);
284 1 : return false;
285 : }
286 :
287 1 : const AICPUSubEventInfo* const subEventInfo = PtrToPtr<const char_t, const AICPUSubEventInfo>(eventInfo.priv.msg);
288 1 : aicpusd_info(
289 : "Begin to process split kernel event. parallelId=%u, threadIdx=%u, type=get",
290 : subEventInfo->para.sharderTaskInfo.parallelId, threadIndex);
291 :
292 1 : return DoSplitKernelTask(subEventInfo->para.sharderTaskInfo);
293 : }
294 :
295 5 : bool ComputeProcess::DoSplitKernelTask(const AICPUSharderTaskInfo& taskInfo)
296 : {
297 5 : aicpu::Closure task;
298 5 : if (!splitKernelTask_.PopTask(taskInfo, task)) {
299 1 : aicpusd_run_warn(
300 : "Get split kernel task from map failed, parallelId=%u, %s", taskInfo.parallelId,
301 : splitKernelTask_.DebugString().c_str());
302 1 : return true;
303 : }
304 :
305 : try {
306 4 : task();
307 1 : } catch (std::exception& e) {
308 1 : aicpusd_err("Run split kernel task failed. parallelId=%u, exception=%s", taskInfo.parallelId, e.what());
309 1 : return false;
310 1 : }
311 :
312 3 : return true;
313 5 : }
314 :
315 3 : bool ComputeProcess::DoRandomKernelTask()
316 : {
317 3 : aicpu::Closure task;
318 3 : if (!randomKernelTask_.Dequeue(task)) {
319 1 : aicpusd_err("Get random kernel task from map failed, %s", randomKernelTask_.DebugString().c_str());
320 1 : return false;
321 : }
322 :
323 : try {
324 2 : task();
325 1 : } catch (std::exception& e) {
326 1 : aicpusd_err("Run random kernel task failed. exception=%s", e.what());
327 1 : return false;
328 1 : }
329 :
330 1 : return true;
331 3 : }
332 :
333 5 : void ComputeProcess::Stop()
334 : {
335 5 : splitKernelTask_.Clear();
336 5 : randomKernelTask_.Clear();
337 5 : if (profilingMode_ == PROFILING_OPEN) {
338 5 : aicpu::ReleaseProfiling();
339 : }
340 5 : }
341 : } // namespace AicpuSchedule
|
