Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 <chrono>
11 : #include <climits>
12 : #include <memory>
13 : #include <atomic>
14 : #include "config_log.h"
15 : #include "launch_device.h"
16 : #include "op_task_config.h"
17 : #include "hccl_common_v2.h"
18 : #include "ascend_hal.h"
19 : #include "orion_adapter_rts.h"
20 : #include "orion_adapter_hal.h"
21 : #include "hccl_dpu_manager.h"
22 :
23 : namespace hccl {
24 : static std::atomic<u32> g_commDpuManagerNum(0);
25 : constexpr uint8_t DEVICE_SIGNAL_SECOND = 2;
26 : constexpr uint8_t DEVICE_SIGNAL_THIRD = 3;
27 : constexpr uint32_t TEMP_DEV_TYPE_DPU = 0; // 临时适配，后续rts接口上库之后使用rts的定义
28 : struct DpuKernelLaunchParam { // 需要和RunDpuRpcSrvLaunch入参定义保持一致
29 : u64 memorySize;
30 : void *deviceMem;
31 : void *hostMem;
32 : int32_t deviceId;
33 : std::string commId;
34 : };
35 : DpuKernelLaunchParam g_hostArgsTemp;
36 :
37 836 : DpuManager::DpuManager()
38 : {
39 836 : }
40 :
41 835 : DpuManager::~DpuManager()
42 : {
43 835 : }
44 :
45 6 : HcclResult DpuManager::CreateWorkspaceBuf(const char *memTag, uint64_t *size, bool *newCreated)
46 : {
47 13 : std::string tag = memTag != nullptr ? std::string(memTag) : "";
48 6 : if (tagWorkspaceMap_.find(tag) == tagWorkspaceMap_.end()) {
49 5 : std::shared_ptr<Hccl::DevBuffer> workspace = std::make_shared<Hccl::DevBuffer>(*size);
50 5 : tagWorkspaceMap_.insert(make_pair(tag, workspace));
51 5 : HCCL_INFO("[DpuManager::%s] Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf = %p", __func__, tag.c_str(),
52 : workspace.get());
53 5 : if (newCreated != nullptr) {
54 3 : *newCreated = true;
55 : }
56 5 : }
57 6 : return HCCL_SUCCESS;
58 6 : }
59 :
60 3 : std::shared_ptr<Hccl::DevBuffer> DpuManager::GetKFCWorkSpace(const char *memTag)
61 : {
62 7 : std::string tag = memTag != nullptr ? std::string(memTag) : "";
63 3 : auto it = tagWorkspaceMap_.find(tag);
64 6 : return it != tagWorkspaceMap_.end() ? it->second : nullptr;
65 3 : }
66 :
67 8 : HcclResult DpuManager::AllocAndRegKFCWorkSpace(uint64_t size)
68 : {
69 8 : accessVA_ = nullptr;
70 8 : drvError_t ret = DRV_ERROR_NONE;
71 :
72 8 : va_ = Hccl::HrtMalloc(size, ACL_MEM_TYPE_HIGH_BAND_WIDTH);
73 8 : ret = halHostRegister(va_, size, DEV_SVM_MAP_HOST, devLogicId_, &accessVA_);
74 8 : if (ret != DRV_ERROR_NONE) {
75 0 : HCCL_ERROR("DpuManager halHostRegister failed, ret: %d", ret);
76 0 : if (va_ != nullptr) {
77 0 : Hccl::HrtFree(va_);
78 : }
79 0 : return HCCL_E_DRV;
80 : }
81 :
82 8 : return HCCL_SUCCESS;
83 : }
84 :
85 0 : HcclResult DpuManager::DestroyKFCWorkSpaceVA()
86 : {
87 0 : if (accessVA_ == nullptr && va_ == nullptr) {
88 0 : HCCL_RUN_WARNING("[%s]accessVA_/va_ is nullptr", __func__);
89 0 : return HCCL_E_INTERNAL;
90 : }
91 :
92 : // 必须先halHostUnregister解除映射，再释放设备内存，否则HrtFree会因内存被pin住而异常
93 0 : if (accessVA_ != nullptr) {
94 0 : drvError_t drvRet = halHostUnregister(accessVA_, devLogicId_);
95 0 : if (drvRet != DRV_ERROR_NONE) {
96 0 : HCCL_ERROR("DpuManager halHostUnregister failed, drvRet[%d]", drvRet);
97 : }
98 : }
99 :
100 0 : if (va_ != nullptr) {
101 0 : Hccl::HrtFree(va_);
102 : }
103 :
104 0 : va_ = nullptr;
105 0 : accessVA_ = nullptr;
106 0 : tagWorkspaceVAMap_.erase(DPUTAG);
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 8 : HcclResult DpuManager::GetKFCWorkSpaceVA(const std::string &memTag, uint64_t *size, void **addr, bool *newCreated)
111 : {
112 8 : if (memTag != DPUTAG) {
113 0 : HCCL_ERROR("DpuManager::GetKFCWorkSpaceVA, memTag is invalid, memTag: %s", memTag.c_str());
114 0 : return HCCL_E_PARA;
115 : }
116 8 : auto iter = tagWorkspaceVAMap_.find(memTag);
117 8 : if (iter != tagWorkspaceVAMap_.end()) {
118 0 : std::shared_ptr<Hccl::DevBuffer> oldWorkspace = iter->second;
119 0 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
120 0 : HCCL_ERROR("DpuManager::GetKFCWorkSpaceVA, The size of oldWorkspace %p is non-consistent, target size compare now size: %llu->%llu", *addr, *size, oldWorkspace.get()->GetSize());
121 0 : return HCCL_E_PARA;
122 : }
123 0 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
124 0 : if (newCreated != nullptr) {
125 0 : *newCreated = false;
126 : }
127 0 : return HcclResult::HCCL_SUCCESS;
128 0 : }
129 :
130 8 : CHK_RET(AllocAndRegKFCWorkSpace(*size));
131 8 : std::shared_ptr<Hccl::DevBuffer> newWorkspace = Hccl::DevBuffer::Create(reinterpret_cast<uintptr_t>(accessVA_), *size);
132 8 : tagWorkspaceVAMap_.insert(make_pair(memTag, newWorkspace));
133 8 : if (newCreated != nullptr) {
134 8 : *newCreated = true;
135 : }
136 8 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
137 8 : return HcclResult::HCCL_SUCCESS;
138 8 : }
139 :
140 6 : HcclResult DpuManager::GetDevMemWorkSpace(const std::string &memTag, uint64_t *size, void **addr, bool *newCreated)
141 : {
142 6 : if (memTag == DPUTAG) {
143 0 : return GetKFCWorkSpaceVA(memTag, size, addr, newCreated);
144 : }
145 6 : auto iter = tagWorkspaceMap_.find(memTag);
146 6 : if (iter != tagWorkspaceMap_.end()) {
147 2 : std::shared_ptr<Hccl::DevBuffer> oldWorkspace = iter->second;
148 2 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
149 1 : HCCL_ERROR("DpuManager::GetDevMemWorkSpace, The size of oldWorkspace %p is non-consistent, "
150 : "target size compare now size: %llu->%llu",
151 : *addr, *size, oldWorkspace.get()->GetSize());
152 1 : return HCCL_E_PARA;
153 : }
154 1 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
155 1 : if (newCreated != nullptr) {
156 1 : *newCreated = false;
157 : }
158 1 : return HcclResult::HCCL_SUCCESS;
159 2 : }
160 :
161 4 : std::shared_ptr<Hccl::DevBuffer> newWorkspace = std::make_shared<Hccl::DevBuffer>(*size);
162 4 : tagWorkspaceMap_.insert(make_pair(memTag, newWorkspace));
163 4 : HCCL_INFO("Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf: %p -> %p, size[%llu]", memTag.c_str(),
164 : newWorkspace.get(), newWorkspace.get()->GetAddr(), *size);
165 4 : if (newCreated != nullptr) {
166 3 : *newCreated = true;
167 : }
168 4 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
169 4 : return HcclResult::HCCL_SUCCESS;
170 4 : }
171 :
172 8 : HcclResult DpuManager::Init(const std::string &commId, u32 deviceLogicId)
173 : {
174 8 : commId_ = commId;
175 8 : devLogicId_ = deviceLogicId;
176 8 : return InitDpuKernel();
177 : }
178 :
179 3 : HcclResult DpuManager::LaunchDpuKernel(aclrtFuncHandle &funcHandle)
180 : {
181 3 : HCCL_INFO("[DpuManager::%s] Launch Dpu Kernel", __func__);
182 : aclrtLaunchKernelCfg cfg;
183 : aclrtLaunchKernelAttr kernelAttr;
184 3 : kernelAttr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
185 3 : kernelAttr.value.timeout = Hccl::NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max()
186 0 : ? std::numeric_limits<uint16_t>::max()
187 : : Hccl::NOTIFY_DEFAULT_WAIT_TIME;
188 3 : cfg.numAttrs = 1;
189 3 : cfg.attrs = &kernelAttr;
190 3 : constexpr u32 numBlocks = 1;
191 :
192 3 : g_hostArgsTemp.commId = commId_;
193 3 : g_hostArgsTemp.memorySize = SHARE_HBM_MEMORY_SIZE;
194 3 : g_hostArgsTemp.hostMem = hostShareBuf_;
195 3 : g_hostArgsTemp.deviceMem = accessVA_;
196 3 : g_hostArgsTemp.deviceId = devLogicId_;
197 :
198 3 : HCCL_INFO("[DpuManager::%s] DpuKernelLaunchParam{commId:%s; memorySize:%u; deviceMem:%p; hostMem:%p}", __func__,
199 : g_hostArgsTemp.commId.c_str(), g_hostArgsTemp.memorySize, g_hostArgsTemp.deviceMem, g_hostArgsTemp.hostMem);
200 :
201 3 : size_t argsSize = sizeof(g_hostArgsTemp);
202 : aclrtPlaceHolderInfo placeHolderArrays;
203 3 : size_t placeHolderNum = 0;
204 3 : if (aclrtLaunchKernelWithHostArgs(
205 : funcHandle, numBlocks, dpuStream_, &cfg, &g_hostArgsTemp, argsSize, &placeHolderArrays, placeHolderNum)
206 3 : != ACL_SUCCESS) {
207 1 : HCCL_ERROR("[DpuManager::%s] Launch Dpu Kernel Failed", __func__);
208 1 : return HCCL_E_INTERNAL;
209 : }
210 2 : return HCCL_SUCCESS;
211 : }
212 :
213 6 : HcclResult DpuManager::PrepareDpuKernelResource(aclrtFuncHandle &funcHandle)
214 : {
215 : // 获取二进制文件路径
216 6 : std::string jsonPath;
217 6 : const char *envPath = getenv("ASCEND_HOME_PATH");
218 6 : if (envPath != nullptr && envPath[0] != '\0') {
219 0 : jsonPath = envPath;
220 : } else {
221 6 : jsonPath = "/usr/local/Ascend/cann/";
222 6 : HCCL_WARNING("[DpuManager::%s] ENV:ASCEND_HOME_PATH is not set", __func__);
223 : }
224 :
225 6 : jsonPath += "/opp/built-in/op_impl/dpu/";
226 6 : HCCL_DEBUG("[DpuManager::%s] kernel folder path[%s]", __func__, jsonPath.c_str());
227 :
228 : // cpuKernelMode为1时，json命名需与so命名保持一致， 即libccl_dpu.json与libccl_dpu.so
229 6 : jsonPath += "libccl_dpu.json";
230 6 : char realPath[PATH_MAX] = {0};
231 6 : CHK_PRT_RET(realpath(jsonPath.c_str(), realPath) == nullptr,
232 : HCCL_ERROR("[DpuManager::%s]: %s is not a valid real path, err[%d]", __func__, jsonPath.c_str(), errno),
233 : HCCL_E_INTERNAL);
234 6 : HCCL_INFO("[DpuManager::%s] realPath: %s", __func__, realPath);
235 :
236 : aclrtBinHandle binHandle;
237 : aclrtBinaryLoadOptions options;
238 : aclrtBinaryLoadOption option;
239 6 : option.type = ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE;
240 6 : option.value.cpuKernelMode = 1;
241 6 : options.numOpt = 1;
242 6 : options.options = &option;
243 6 : if (aclrtBinaryLoadFromFile(realPath, &options, &binHandle) != ACL_SUCCESS) {
244 1 : HCCL_ERROR("[DpuManager::%s] load binary from file error.", __func__);
245 1 : return HCCL_E_OPEN_FILE_FAILURE;
246 : }
247 :
248 : // 创建dpustream
249 5 : if (aclrtCreateStreamWithConfig(&dpuStream_, 0, ACL_STREAM_FAST_LAUNCH) != ACL_SUCCESS) {
250 1 : HCCL_ERROR("[DpuManager::%s] Create Local Stream Failed", __func__);
251 1 : return HCCL_E_INTERNAL;
252 : }
253 :
254 : // 查找核函数
255 4 : if (aclrtBinaryGetFunction(binHandle, "RunDpuRpcSrvLaunch", &funcHandle) != ACL_SUCCESS) {
256 1 : HCCL_ERROR("[DpuManager::%s] Get Function Failed", __func__);
257 1 : return HCCL_E_INTERNAL;
258 : }
259 :
260 3 : return HCCL_SUCCESS;
261 6 : }
262 :
263 8 : HcclResult DpuManager::InitAndLaunchDpuKernel()
264 : {
265 8 : HCCL_INFO("[DpuManager::%s] Start to Launch Dpu Kernel", __func__);
266 8 : bool newCreate = false;
267 8 : uint64_t memSize = static_cast<uint64_t>(SHARE_HBM_MEMORY_SIZE);
268 16 : HcclResult memRet = GetKFCWorkSpaceVA(DPUTAG, &memSize, &accessVA_, &newCreate);
269 8 : if (memRet != HCCL_SUCCESS) {
270 0 : HCCL_ERROR("[DpuManager::InitCommResource] Alloc Share HBM Failed");
271 0 : return HCCL_E_RUNTIME;
272 : }
273 :
274 : // 设置XPU
275 8 : HCCL_INFO("[DpuManager::%s] Switch to Dpu Ctx", __func__);
276 8 : if (aclrtGetCurrentContext(&npuContext_) != ACL_SUCCESS) {
277 1 : HCCL_ERROR("[DpuManager::%s] Get Npu Ctx Failed", __func__);
278 1 : return HCCL_E_INTERNAL;
279 : }
280 7 : if (Hccl::HrtSetXpuDevice(TEMP_DEV_TYPE_DPU, 0) != HCCL_SUCCESS) {
281 1 : HCCL_ERROR("[DpuManager::%s] Switch to Dpu Ctx Failed", __func__);
282 1 : return HCCL_E_INTERNAL;
283 : }
284 6 : if (aclrtGetCurrentContext(&dpuContext_) != ACL_SUCCESS) {
285 0 : HCCL_ERROR("[DpuManager::%s] Get Dpu Ctx Failed", __func__);
286 0 : return HCCL_E_INTERNAL;
287 : }
288 :
289 : // 准备资源
290 : aclrtFuncHandle funcHandle;
291 6 : CHK_RET(PrepareDpuKernelResource(funcHandle));
292 :
293 3 : hostShareBuf_ = malloc(SHARE_HBM_MEMORY_SIZE);
294 3 : CHK_PTR_NULL(hostShareBuf_);
295 :
296 : // 下发
297 3 : HcclResult ret = LaunchDpuKernel(funcHandle);
298 3 : if (ret != HCCL_SUCCESS) {
299 1 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Dpu Kernel Failed", __func__);
300 1 : free(hostShareBuf_);
301 1 : hostShareBuf_ = nullptr;
302 1 : return ret;
303 : }
304 :
305 : // 切换回当前Ctx
306 2 : HCCL_INFO("[DpuManager::%s] Switch to Npu Ctx", __func__);
307 2 : if (ACL_SUCCESS != aclrtSetCurrentContext(npuContext_)) {
308 1 : HCCL_ERROR("[DpuManager::%s] Reset Current Ctx Failed", __func__);
309 1 : free(hostShareBuf_);
310 1 : hostShareBuf_ = nullptr;
311 1 : return HCCL_E_INTERNAL;
312 : }
313 :
314 1 : HCCL_INFO("[DpuManager::%s] Launch Dpu Kernel End", __func__);
315 1 : isDpuKernelLaunched_ = true;
316 1 : g_commDpuManagerNum++;
317 1 : return HCCL_SUCCESS;
318 : }
319 :
320 8 : HcclResult DpuManager::InitDpuKernel()
321 : {
322 : /* kernel Launch */
323 8 : CHK_RET(InitAndLaunchDpuKernel());
324 1 : return HCCL_SUCCESS;
325 : }
326 :
327 805 : HcclResult DpuManager::DeInitDpuKernel()
328 : {
329 805 : if (!isDpuKernelLaunched_) {
330 806 : return HCCL_SUCCESS;
331 : }
332 0 : (void)DestroyDpuKernelResource();
333 0 : (void)DestroyKFCWorkSpaceVA();
334 0 : if (hostShareBuf_ != nullptr) {
335 0 : free(hostShareBuf_);
336 0 : hostShareBuf_ = nullptr;
337 : }
338 0 : return HCCL_SUCCESS;
339 : }
340 :
341 3 : HcclResult DpuManager::WaitDpuKernelThreadTerminate()
342 : {
343 3 : if (!isDpuKernelLaunched_) {
344 1 : return HCCL_SUCCESS;
345 : }
346 2 : if (accessVA_ == nullptr) {
347 1 : HCCL_ERROR("[CommunicatorImpl::%s] accessVA_ is nullptr", __func__);
348 1 : return HCCL_E_MEMORY;
349 : }
350 1 : uint8_t flag = DEVICE_SIGNAL_SECOND;
351 1 : errno_t ret = memcpy_s(accessVA_, sizeof(flag), &flag, sizeof(flag));
352 1 : if (ret != EOK) {
353 0 : HCCL_ERROR("Terminate TaskRun Fail, return[%d]", ret);
354 0 : return HCCL_E_MEMORY;
355 : }
356 : do {
357 2859 : ret = memcpy_s(&flag, sizeof(flag), accessVA_, sizeof(flag));
358 2859 : if (ret != EOK) {
359 0 : HCCL_ERROR("Read Terminate TaskRun Signal Fail, return[%d]", ret);
360 0 : return HCCL_E_MEMORY;
361 : }
362 2859 : } while (flag != DEVICE_SIGNAL_THIRD);
363 :
364 1 : return HCCL_SUCCESS;
365 : }
366 :
367 :
368 4 : HcclResult DpuManager::DestroyDpuKernelResource()
369 : {
370 : // 终止Dpu Kernel的TaskRun
371 4 : if (!isDpuKernelLaunched_) {
372 0 : return HCCL_SUCCESS;
373 : }
374 :
375 4 : CHK_RET(WaitDpuKernelThreadTerminate());
376 :
377 : // 切换回 dpu ctx
378 4 : aclError aclRet = aclrtSetCurrentContext(dpuContext_);
379 4 : if (ACL_SUCCESS != aclRet) {
380 1 : HCCL_ERROR("set dpu Ctx Failed, aclReturn[%d]", aclRet);
381 1 : return HCCL_E_RUNTIME;
382 : }
383 : // 销毁局部流
384 3 : aclRet = aclrtDestroyStreamForce(dpuStream_);
385 3 : if (ACL_SUCCESS != aclRet) {
386 1 : HCCL_ERROR("Destroy Stream Failed, aclReturn[%d]", aclRet);
387 1 : aclRet = aclrtSetCurrentContext(npuContext_);
388 1 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
389 0 : return HCCL_E_RUNTIME;
390 : }
391 2 : if (g_commDpuManagerNum > 1) {
392 0 : g_commDpuManagerNum--;
393 : } else {
394 : // reset DPU kernel 线程
395 2 : HcclResult ret = Hccl::HrtResetXpuDevice(TEMP_DEV_TYPE_DPU, 0);
396 2 : if (HCCL_SUCCESS != ret) {
397 1 : HCCL_ERROR("ResetXpuDevice Failed, return[%d]", ret);
398 1 : aclRet = aclrtSetCurrentContext(npuContext_);
399 1 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
400 0 : return HCCL_E_RUNTIME;
401 : }
402 : }
403 : // 切回 npu ctx
404 1 : aclRet = aclrtSetCurrentContext(npuContext_);
405 1 : if (ACL_SUCCESS != aclRet) {
406 0 : HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet);
407 0 : return HCCL_E_RUNTIME;
408 : }
409 :
410 1 : return HCCL_SUCCESS;
411 : }
412 :
413 : } // namespace hccl
|
