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