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