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