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 "task_service.h"
12 : #include <algorithm>
13 : #include "profiling/dlprof_function_v2.h"
14 : #include <thread>
15 : #include <chrono>
16 : #include <atomic>
17 : #include "acl/acl_rt.h"
18 : #include "log.h"
19 : #include "dpu_kernel_entrance.h"
20 : #include "env_config/env_config_v2.h"
21 :
22 : namespace Hccl {
23 : constexpr uint32_t CTRL_HDR_FLAG_LENGTH = 1;
24 : constexpr uint32_t TASKTYPE_ADDR_LENGTH = 256;
25 : constexpr uint32_t CTRL_HDR_MSG_ID_LEN = 4;
26 : constexpr uint32_t TIMEOUT_SIZE_BYTE = 4; // timeout字段长度为4字节，表示超时时间，单位为秒
27 : constexpr uint32_t CTRL_HDR_DATA_SIZE_LEN = 8; // size_t 在不同平台上长度不同，取最大值
28 : constexpr uint32_t CTRL_HDR_DEFAULT_DATA_LEN = 512;
29 :
30 : constexpr uint8_t TASK_UNSET = 0;
31 : constexpr uint8_t TASK_OK = 1;
32 : constexpr uint8_t TASK_TERMINATE = 2;
33 : constexpr uint8_t MEMORY_DEVIDE = 2;
34 :
35 25 : TaskService::TaskService(
36 25 : void* deviceMem, int32_t deviceMemSize, void* hostMem, int32_t hostMemSize, std::string commId, uint32_t devId)
37 25 : : npu2dpuMem_(deviceMem),
38 25 : shmemSize_(deviceMemSize / MEMORY_DEVIDE),
39 25 : hostMem_(hostMem),
40 25 : hostMemSize_(hostMemSize),
41 25 : commId_(commId),
42 25 : devId_(devId)
43 : {
44 25 : int32_t controlSize = sizeof(uint8_t) + sizeof(char) * TASKTYPE_ADDR_LENGTH + sizeof(uint32_t) + TIMEOUT_SIZE_BYTE
45 : + CTRL_HDR_DATA_SIZE_LEN;
46 25 : if (shmemSize_ < controlSize) {
47 1 : leftSize_ = 0;
48 : } else {
49 24 : leftSize_ = shmemSize_ - controlSize;
50 : }
51 25 : dpu2npuMem_ = static_cast<uint8_t*>(npu2dpuMem_) + shmemSize_;
52 25 : }
53 :
54 9 : HcclResult TaskService::TaskRegister(std::string taskType, CallbackTemplate callback)
55 : {
56 27 : HCCL_INFO("[TaskService::%s] taskType[%s]", __func__, taskType.c_str());
57 9 : callbacks_.insert({taskType, callback});
58 9 : return HCCL_SUCCESS;
59 : }
60 :
61 2 : HcclResult TaskService::TaskUnRegister(std::string taskType)
62 : {
63 6 : HCCL_INFO("[TaskService::%s] taskType[%s]", __func__, taskType.c_str());
64 2 : if (callbacks_.find(taskType) == callbacks_.end()) {
65 3 : HCCL_WARNING("[TaskService::%s] TaskType Not Found", __func__);
66 1 : return HCCL_E_NOT_FOUND;
67 : }
68 1 : callbacks_.erase(taskType);
69 1 : return HCCL_SUCCESS;
70 : }
71 :
72 5 : HcclResult TaskService::TaskProfRegister(ProfCallbackTemplate profCallback)
73 : {
74 5 : if (profCallback_ != nullptr) {
75 1 : return HCCL_SUCCESS;
76 : }
77 4 : profCallback_ = profCallback;
78 4 : return HCCL_SUCCESS;
79 : }
80 :
81 4 : HcclResult TaskService::TaskReportRegister(ReportCallbackTemplate reportCallback)
82 : {
83 4 : if (reportCallback_ != nullptr) {
84 1 : return HCCL_SUCCESS;
85 : }
86 3 : reportCallback_ = reportCallback;
87 3 : return HCCL_SUCCESS;
88 : }
89 :
90 9 : HcclResult TaskService::WriteFlag(uint8_t* flagPtr, uint8_t newFlag) const
91 : {
92 9 : errno_t ret = memcpy_s(flagPtr, sizeof(newFlag), &newFlag, sizeof(newFlag));
93 9 : if (ret != EOK) {
94 0 : HCCL_ERROR("[TaskService::TaskRun] set flag failed: %d", ret);
95 0 : return HCCL_E_INTERNAL;
96 : }
97 9 : return HCCL_SUCCESS;
98 : }
99 :
100 12 : HcclResult TaskService::ReadFlag(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t& flag) const
101 : {
102 12 : errno_t ret = memcpy_s(ctrlHdr, hdrLen, npu2dpuMem_, hdrLen);
103 12 : if (ret != EOK) {
104 0 : HCCL_ERROR("[TaskService::%s] memcpy_s failed on flag, return[%d].", __func__, ret);
105 0 : return HCCL_E_INTERNAL;
106 : }
107 12 : flag = *ctrlHdr;
108 12 : return HCCL_SUCCESS;
109 : }
110 :
111 7 : HcclResult TaskService::ReadTaskType(
112 : const uint8_t* ctrlHdr, [[maybe_unused]] uint64_t hdrLen, const uint8_t* srcTaskTypePtr,
113 : std::string& taskTypeStr) const
114 : {
115 7 : CHK_PTR_NULL(srcTaskTypePtr);
116 : // 读 taskType
117 7 : char* taskType = new char[TASKTYPE_ADDR_LENGTH];
118 7 : CHK_PTR_NULL(taskType);
119 14 : int ret = memcpy_s(
120 7 : taskType, (sizeof(char) * TASKTYPE_ADDR_LENGTH), ctrlHdr + CTRL_HDR_FLAG_LENGTH,
121 : (sizeof(char) * TASKTYPE_ADDR_LENGTH));
122 7 : if (ret != EOK) {
123 0 : HCCL_ERROR("[%s] memcpy failed on taskType, return[%d].", __func__, ret);
124 0 : delete[] taskType;
125 0 : return HCCL_E_INTERNAL;
126 : }
127 : // 查找 \0
128 7 : auto it = std::find(taskType, taskType + TASKTYPE_ADDR_LENGTH, '\0');
129 7 : if (it == taskType + TASKTYPE_ADDR_LENGTH) {
130 0 : HCCL_ERROR("[TaskService::TaskRun] No Null Character Found Within TaskType Max Length");
131 0 : delete[] taskType;
132 0 : return HCCL_E_PARA;
133 : }
134 7 : taskTypeStr = std::string(taskType);
135 7 : delete[] taskType;
136 21 : HCCL_INFO("[TaskService::TaskRun] read taskType = %s", taskTypeStr.c_str());
137 :
138 7 : uint32_t msgId{0};
139 7 : ret = memcpy_s(&msgId, sizeof(msgId), ctrlHdr + CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH, sizeof(msgId));
140 7 : if (ret != EOK) {
141 0 : HCCL_ERROR("[%s] memcpy failed on msgId, return[%d].", __func__, ret);
142 0 : return HCCL_E_INTERNAL;
143 : }
144 21 : HCCL_INFO("[TaskService::TaskRun] read msgId = %u.", msgId);
145 7 : return HCCL_SUCCESS;
146 : }
147 :
148 : // 共享内存排布：|stop flag[1]|hcclret[2]|dstret[2]|
149 : // 其中，stop flag为aicpu侧读取是否停止的标志位, hcclret为aicpu背景线程读取是否有错的标志位,
150 : // dstret为host侧taskexception回调读取是否有错的标志位
151 7 : HcclResult TaskService::ExecuteTaskexception(int32_t ret)
152 : {
153 7 : if (g_taskExpMemMap.find(commId_) == g_taskExpMemMap.end()) {
154 3 : HCCL_ERROR("TaskService::ExecuteTaskexception commId not in g_taskExpMemMap, please check");
155 1 : return HCCL_E_NOT_FOUND;
156 : }
157 6 : void* taskexpShmem = g_taskExpMemMap[commId_][devId_];
158 6 : HcclResult hcclRet = static_cast<HcclResult>(ret);
159 6 : if (taskexpShmem != nullptr) {
160 4 : uint8_t* stopFlagPtr = static_cast<uint8_t*>(taskexpShmem);
161 4 : uint8_t* hcclRetPtr = stopFlagPtr + sizeof(uint8_t);
162 4 : uint8_t* dstRetPtr = hcclRetPtr + sizeof(uint16_t);
163 4 : auto ret = memcpy_s(hcclRetPtr, sizeof(uint16_t), &hcclRet, sizeof(uint16_t)); // aicpu背景线程轮询的标志位
164 4 : if (ret != 0) {
165 0 : HCCL_ERROR("[TaskService::ExecuteTaskexception] memcpy ret for device failed.");
166 0 : return HCCL_E_MEMORY;
167 : }
168 4 : ret = memcpy_s(dstRetPtr, sizeof(uint16_t), &hcclRet, sizeof(uint16_t)); // host侧taskexception回调执读出错误码
169 4 : if (ret != 0) {
170 0 : HCCL_ERROR("[TaskService::ExecuteTaskexception] memcpy ret for host failed.");
171 0 : return HCCL_E_MEMORY;
172 : }
173 : }
174 6 : return HCCL_SUCCESS;
175 : }
176 :
177 5 : HcclResult TaskService::ExecuteTaskClean() const
178 : {
179 5 : errno_t ret = memset_s(hostMem_, hostMemSize_, 0, hostMemSize_);
180 5 : if (ret != EOK) {
181 0 : HCCL_ERROR("memset hostMem failed: %d", ret);
182 0 : return HCCL_E_MEMORY;
183 : }
184 5 : ret = memset_s(npu2dpuMem_, shmemSize_, 0, shmemSize_);
185 5 : if (ret != EOK) {
186 0 : HCCL_ERROR("memset npu2dpuMem_ failed: %d", ret);
187 0 : return HCCL_E_MEMORY;
188 : }
189 5 : ret = memset_s(dpu2npuMem_, shmemSize_, 0, shmemSize_);
190 5 : if (ret != EOK) {
191 0 : HCCL_ERROR("memset dpu2npuMem_ failed: %d", ret);
192 0 : return HCCL_E_MEMORY;
193 : }
194 5 : return HCCL_SUCCESS;
195 : }
196 :
197 7 : HcclResult TaskService::ExecuteTask(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t* srcPtr, std::string taskTypeStr)
198 : {
199 7 : uint64_t beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
200 7 : auto itFunc = callbacks_.find(taskTypeStr);
201 7 : if (itFunc == callbacks_.end()) {
202 3 : HCCL_ERROR("[TaskService::TaskRun] Callback of taskType[%s] Not Found", taskTypeStr.c_str());
203 1 : std::string taskTypeMsg{"map{"};
204 2 : for (const auto& pair : callbacks_) {
205 1 : taskTypeMsg += pair.first;
206 : }
207 1 : taskTypeMsg += "}";
208 3 : HCCL_ERROR("[TaskService::TaskRun] Callback key : %s", taskTypeMsg.c_str());
209 1 : return HCCL_E_NOT_FOUND;
210 1 : }
211 :
212 : // copy data
213 6 : uint64_t dataLen = *reinterpret_cast<size_t*>(
214 : ctrlHdr + CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE);
215 6 : if (dataLen > static_cast<uint64_t>(leftSize_) || dataLen > static_cast<uint64_t>(hostMemSize_)) {
216 3 : HCCL_ERROR(
217 : "[TaskService::%s] dataLen[%llu] larger than leftSize[%d] or hostMemSize[%d]", __func__, dataLen, leftSize_,
218 : hostMemSize_);
219 1 : return HCCL_E_PARA;
220 : }
221 5 : uint32_t ctrlHdrLen = CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE
222 : + CTRL_HDR_DATA_SIZE_LEN;
223 : /* ctrlHdr提前从deviceMem copy一定长度，如果长度够，直接从ctrlHdr copy，减少一次aclmemcpy耗时 */
224 5 : uint8_t* dataPtr = nullptr;
225 5 : if (hdrLen < ctrlHdrLen + dataLen) {
226 0 : dataPtr = srcPtr + ctrlHdrLen;
227 : } else {
228 5 : dataPtr = ctrlHdr + ctrlHdrLen;
229 : }
230 5 : errno_t ret = memcpy_s(hostMem_, leftSize_, dataPtr, dataLen);
231 5 : if (ret != EOK) {
232 0 : HCCL_ERROR("control data memcpy failed: %d", ret);
233 0 : return HCCL_E_MEMORY;
234 : }
235 5 : auto callbackRet = itFunc->second(reinterpret_cast<uint64_t>(hostMem_), dataLen);
236 5 : if (callbackRet != 0) {
237 : // dpu任务出错，清理DPUTAG共享内存内容
238 4 : CHK_RET(ExecuteTaskClean());
239 : // 写DPUTASKEXCEPTION
240 4 : CHK_RET(ExecuteTaskexception(callbackRet));
241 4 : return HCCL_E_INTERNAL;
242 : }
243 1 : if (profCallback_ != nullptr) {
244 0 : TaskParam taskParam{};
245 0 : taskParam.beginTime = beginTime;
246 0 : taskParam.taskType = Hccl::TaskParamType::TASK_DPU_KERNEL;
247 0 : taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
248 0 : taskParam.isMaster = true;
249 0 : profCallback_(taskParam, INVALID_U64);
250 0 : }
251 1 : return HCCL_SUCCESS;
252 : }
253 :
254 1 : HcclResult TaskService::SynchronizeControlInfo([[maybe_unused]] uint8_t* ctrlHdr, [[maybe_unused]] uint64_t hdrLen)
255 : {
256 1 : CHK_PTR_NULL(npu2dpuMem_);
257 1 : CHK_PTR_NULL(dpu2npuMem_);
258 : // npu2dpu -> dpu2npu
259 1 : int32_t controlDataSize = sizeof(uint8_t) + sizeof(char) * TASKTYPE_ADDR_LENGTH + sizeof(uint32_t);
260 3 : HCCL_INFO("[TaskService::TaskRun] Send response: npu2dpu -> dpu2npu memcpy");
261 1 : errno_t ret = memcpy_s(dpu2npuMem_, controlDataSize, npu2dpuMem_, controlDataSize);
262 1 : if (ret != EOK) {
263 0 : HCCL_ERROR("[TaskService::TaskRun] npu2dpu -> dpu2npu memcpy failed: %d", ret);
264 0 : return HCCL_E_INTERNAL;
265 : }
266 :
267 : std::atomic_thread_fence(std::memory_order_seq_cst);
268 :
269 1 : uint8_t newFlag = 1;
270 3 : HCCL_INFO("[TaskService::TaskRun] Send response: Set dpu2npu flag -> 1");
271 1 : static_cast<std::atomic<uint8_t>*>(dpu2npuMem_)->store(newFlag, std::memory_order_release);
272 1 : return HCCL_SUCCESS;
273 : }
274 :
275 7 : HcclResult TaskService::ProcessTaskOk(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t* srcFlagPtr, uint8_t* srcTaskTypePtr)
276 : {
277 7 : std::string taskTypeStr;
278 21 : HCCL_INFO("[TaskService::TaskRun] flag = %u.", TASK_OK);
279 21 : HCCL_INFO("[TaskService::TaskRun] Set npu2dpu flag -> %u.", TASK_UNSET);
280 7 : CHK_RET(WriteFlag(srcFlagPtr, TASK_UNSET));
281 7 : CHK_RET(ReadTaskType(ctrlHdr, hdrLen, srcTaskTypePtr, taskTypeStr));
282 25 : CHK_RET(ExecuteTask(ctrlHdr, hdrLen, srcFlagPtr, taskTypeStr));
283 1 : if (reportCallback_ != nullptr) {
284 0 : CHK_RET(reportCallback_());
285 : }
286 1 : CHK_RET(SynchronizeControlInfo(ctrlHdr, hdrLen));
287 1 : return HCCL_SUCCESS;
288 7 : }
289 :
290 2 : HcclResult TaskService::ExecuteExit(uint8_t* srcFlagPtr) const
291 : {
292 2 : CHK_RET(WriteFlag(srcFlagPtr, TASK_TERMINATE_RESPONSE));
293 6 : HCCL_INFO("[TaskService::TaskRun] Exiting.");
294 2 : return HCCL_SUCCESS;
295 : }
296 :
297 12 : HcclResult TaskService::TaskRun()
298 : {
299 0 : CHK_PTR_NULL(hostMem_);
300 12 : CHK_PTR_NULL(npu2dpuMem_);
301 12 : CHK_PTR_NULL(dpu2npuMem_);
302 36 : HCCL_INFO(
303 : "[TaskService::%s] TaskService{npu2dpuMem:%p; dpu2npuMem:%p; hostMem:%p}", __func__, npu2dpuMem_, dpu2npuMem_,
304 : hostMem_);
305 12 : if (leftSize_ <= 0) {
306 3 : HCCL_ERROR("[TaskService::%s] dataSize[%d] illegal", __func__, leftSize_);
307 1 : return HCCL_E_INTERNAL;
308 : }
309 11 : if (leftSize_ > hostMemSize_) {
310 3 : HCCL_ERROR("[TaskService::%s] hostMemSize[%d] less than dataSize[%d]", __func__, hostMemSize_, leftSize_);
311 1 : return HCCL_E_INTERNAL;
312 : }
313 10 : uint8_t flag{0};
314 10 : uint8_t* srcFlagPtr = static_cast<uint8_t*>(npu2dpuMem_);
315 10 : uint8_t* dstTimeoutPtr
316 10 : = static_cast<uint8_t*>(dpu2npuMem_) + sizeof(flag) + sizeof(char) * TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN;
317 10 : uint8_t* srcTaskTypePtr = srcFlagPtr + sizeof(flag);
318 10 : uint64_t hdrLen = CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE
319 : + CTRL_HDR_DATA_SIZE_LEN + CTRL_HDR_DEFAULT_DATA_LEN;
320 10 : uint8_t ctrlHdr[hdrLen];
321 10 : u32 timeout = Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
322 10 : timeout = timeout > 1 ? timeout - 1 : timeout; // 执行超时时间减1秒，避免aicpu侧超时
323 10 : CHK_SAFETY_FUNC_RET(memcpy_s(dstTimeoutPtr, sizeof(timeout), &timeout, sizeof(timeout)));
324 :
325 : while (true) {
326 12 : CHK_RET(ReadFlag(ctrlHdr, hdrLen, flag));
327 12 : switch (flag) {
328 1 : case TASK_UNSET:
329 1 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
330 1 : continue;
331 7 : case TASK_OK:
332 25 : CHK_RET(ProcessTaskOk(ctrlHdr, hdrLen, srcFlagPtr, srcTaskTypePtr));
333 1 : continue;
334 2 : case TASK_TERMINATE:
335 6 : HCCL_INFO("[TaskService::TaskRun] flag = %u.", flag);
336 6 : HCCL_INFO("[TaskService::TaskRun] Set npu2dpu flag -> %u. Task Run END.", TASK_TERMINATE_RESPONSE);
337 2 : CHK_RET(ExecuteExit(srcFlagPtr));
338 2 : return HCCL_SUCCESS;
339 1 : case TASK_TERMINATE_RESPONSE:
340 3 : HCCL_INFO("[TaskService::TaskRun] flag = %u. Exiting.", flag);
341 1 : return HCCL_SUCCESS;
342 1 : default:
343 3 : HCCL_INFO("[TaskService::TaskRun] flag = %u. Unimplemented flag. Exiting.", flag);
344 1 : return HCCL_SUCCESS;
345 2 : }
346 2 : }
347 : return HCCL_SUCCESS;
348 12 : }
349 : } // namespace Hccl
|
