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 "hccl_one_sided_service.h"
12 : #include <future>
13 : #include "device_capacity.h"
14 : #include "sal_pub.h"
15 : #include "threads_guard.h"
16 : #include "adapter_rts_common.h"
17 : #include "adapter_prof.h"
18 : #include "profiling_manager_pub.h"
19 : #include "prof_common.h"
20 : #include "launch_aicpu.h"
21 : #include "launch_device.h"
22 : #include "comm_configer.h"
23 :
24 : namespace hccl {
25 : using namespace std;
26 : constexpr u32 INVALID_REMOTE_RANK_ID = 0xFFFFFFFF;
27 : constexpr u64 TILINGDATA_BUF_SIZE = 32 * 1024;
28 : constexpr u16 MAX_VALUE_U16 = 0xFFFF;
29 :
30 : std::mutex HcclOneSidedService::regMutex_;
31 :
32 : std::unique_ptr<Stream> g_launchStream = nullptr;
33 : std::mutex g_launchMutex;
34 :
35 504 : HcclOneSidedService::HcclOneSidedService(
36 504 : unique_ptr<HcclSocketManager>& socketManager, unique_ptr<NotifyPool>& notifyPool, const CommConfig& commConfig)
37 504 : : IHcclOneSidedService(socketManager, notifyPool)
38 : {
39 504 : commConfig_ = commConfig;
40 504 : }
41 :
42 1005 : HcclOneSidedService::~HcclOneSidedService()
43 : {
44 504 : HCCL_RUN_INFO(
45 : "[~HcclOneSidedService] localRankId[%u] has registedMemCnt[%u] mem didn't dereg", localRankInfo_.userRank,
46 : registedMemCnt_);
47 504 : HcclResult ret = HCCL_SUCCESS;
48 504 : for (auto it = desc2HcclBufMapIpc_.begin(); it != desc2HcclBufMapIpc_.end(); ++it) {
49 0 : HcclBuf& buf = it->second;
50 : do {
51 0 : ret = HcclMemDereg(&buf); // 需循环调用DeregMem来去注册内存(因为存在一块内存多次Reg的情况)
52 : // 失败场景记录log即可，接着处理后面的mem
53 0 : CHK_PRT_CONT(
54 : ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)),
55 : HCCL_ERROR(
56 : "[~HcclOneSidedService] DeregMem IPC localRankId[%u] addr[%p] size[%lu] failed",
57 : localRankInfo_.userRank, buf.addr, buf.len));
58 0 : } while (ret == HCCL_E_AGAIN);
59 : }
60 :
61 504 : for (auto it = desc2HcclBufMapRoce_.begin(); it != desc2HcclBufMapRoce_.end(); ++it) {
62 0 : HcclBuf& buf = it->second;
63 : do {
64 0 : ret = HcclMemDereg(&buf); // 需循环调用DeregMem来去注册内存(因为存在一块内存多次Reg的情况)
65 : // 失败场景记录log即可，接着处理后面的mem
66 0 : CHK_PRT_CONT(
67 : ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)),
68 : HCCL_ERROR(
69 : "[~HcclOneSidedService] DeregMem ROCE localRankId[%u] addr[%p] size[%lu] failed",
70 : localRankInfo_.userRank, buf.addr, buf.len));
71 0 : } while (ret == HCCL_E_AGAIN);
72 : }
73 :
74 3022 : for (u32 i = 0; i < localAicpuNotify_.size(); ++i) {
75 1008 : if (localAicpuNotify_[i] != nullptr) {
76 0 : ret = localAicpuNotify_[i]->Destroy();
77 0 : localAicpuNotify_[i] = nullptr;
78 0 : if (ret != HCCL_SUCCESS) {
79 0 : HCCL_ERROR(
80 : "[Destroy][AiCpuNotify] errNo[0x%016llx] notify destroy fail, aicpuNotify[%u], ret[%d].",
81 : HCCL_ERROR_CODE(HCCL_E_RUNTIME), i, ret);
82 : }
83 : }
84 : }
85 503 : UnloadAICPUKernel();
86 1005 : }
87 :
88 0 : HcclResult HcclOneSidedService::IsUsedRdma(RankId remoteRankId, bool& useRdma)
89 : {
90 : DevType deviceType;
91 0 : CHK_RET(hrtGetDeviceType(deviceType));
92 :
93 0 : RankInfo_t localRankInfo = (rankTable_->rankList).at(localRankInfo_.userRank);
94 0 : RankInfo_t remoteRankInfo = (rankTable_->rankList).at(remoteRankId);
95 0 : if (deviceType == DevType::DEV_TYPE_910B) {
96 : // 外部使能RDMA，或者节点间通信
97 0 : if (GetExternalInputIntraRoceSwitch() != 0 || localRankInfo.serverId != remoteRankInfo.serverId) {
98 0 : useRdma = true;
99 0 : return HCCL_SUCCESS;
100 : }
101 :
102 : // 同一节点的 PCIe 连接判断
103 0 : s32 localDeviceId = localRankInfo_.devicePhyId;
104 0 : s32 remoteDeviceId = remoteRankInfo.deviceInfo.devicePhyId;
105 0 : LinkTypeInServer linkType = LinkTypeInServer::RESERVED_LINK_TYPE;
106 0 : CHK_RET(hrtGetPairDeviceLinkType(static_cast<u32>(localDeviceId), static_cast<u32>(remoteDeviceId), linkType));
107 0 : if (linkType != LinkTypeInServer::HCCS_TYPE) {
108 0 : HCCL_ERROR(
109 : "[HcclOneSidedService][IsUsedRdma]localDeviceId: %d, remoteDeviceId: %d, linkType %u is not supported",
110 : localDeviceId, remoteDeviceId, linkType);
111 0 : return HCCL_E_NOT_SUPPORT;
112 : }
113 :
114 : // 节点内通信，默认不使用 RDMA
115 0 : useRdma = false;
116 0 : return HCCL_SUCCESS;
117 0 : } else if (deviceType == DevType::DEV_TYPE_910_93) {
118 0 : if (GetExternalInputIntraRoceSwitch() != 0 || localRankInfo.superPodId != remoteRankInfo.superPodId) {
119 0 : useRdma = true;
120 0 : return HCCL_SUCCESS;
121 : }
122 :
123 0 : useRdma = false;
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 : // 其他情况默认使用 RDMA
128 0 : useRdma = true;
129 0 : return HCCL_SUCCESS;
130 0 : }
131 :
132 0 : HcclResult HcclOneSidedService::GetIsUsedRdma(RankId remoteRankId, bool& useRdma)
133 : {
134 0 : if (isUsedRdmaMap_.find(remoteRankId) == isUsedRdmaMap_.end()) {
135 0 : CHK_RET(IsUsedRdma(remoteRankId, useRdma));
136 0 : isUsedRdmaMap_[remoteRankId] = useRdma;
137 : } else {
138 0 : useRdma = isUsedRdmaMap_[remoteRankId];
139 : }
140 :
141 0 : return HCCL_SUCCESS;
142 : }
143 :
144 1 : HcclResult HcclOneSidedService::ReMapMem(HcclMem* memInfoArray, u64 arraySize)
145 : {
146 1 : HcclResult ret = HCCL_SUCCESS;
147 1 : if (netDevRdmaCtx_) { // 非roce场景不进行remap，返回success
148 0 : ret = HcclMemRemap(netDevRdmaCtx_, memInfoArray, arraySize);
149 : } else {
150 1 : HCCL_RUN_INFO("[HcclOneSidedService][ReMapMem] doesn't support remap ipc mem, just return success");
151 : }
152 1 : return ret;
153 : }
154 :
155 : HcclResult
156 0 : HcclOneSidedService::RegMem(void* addr, u64 size, HcclMemType type, RankId remoteRankId, HcclMemDesc& localMemDesc)
157 : {
158 0 : bool useRdma = true;
159 0 : if (isUsedRdmaMap_.find(remoteRankId) == isUsedRdmaMap_.end()) {
160 0 : CHK_RET(IsUsedRdma(remoteRankId, useRdma));
161 0 : isUsedRdmaMap_[remoteRankId] = useRdma;
162 : }
163 0 : useRdma = isUsedRdmaMap_[remoteRankId];
164 :
165 0 : HcclMem localMem{type, addr, size};
166 : HcclBuf buf;
167 0 : HcclResult ret = HcclMemReg(useRdma ? netDevRdmaCtx_ : netDevIpcCtx_, &localMem, &buf);
168 0 : if ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)) { // HCCL_E_AGAIN:调用HcclMemReg前，内存已注册过
169 0 : return ret;
170 : }
171 0 : bool firstReg = (ret == HCCL_SUCCESS);
172 :
173 0 : char* desc = nullptr;
174 0 : uint64_t descLen = 0;
175 0 : ret = HcclMemExport(&buf, &desc, &descLen);
176 0 : if (ret != HCCL_SUCCESS) {
177 0 : HCCL_ERROR("[HcclOneSidedService][RegMem] get mem desc failed, ret[%d]", ret);
178 0 : throw logic_error("[HcclOneSidedService][RegMem] get mem desc failed");
179 : }
180 :
181 0 : HcclMemDescData* ptr = static_cast<HcclMemDescData*>(static_cast<void*>(localMemDesc.desc));
182 0 : ptr->localRankId = localRankInfo_.userRank;
183 0 : ptr->remoteRankId = remoteRankId;
184 0 : memset_s(ptr->memDesc, HCCL_MEM_DESC_STR_LEN, 0, HCCL_MEM_DESC_STR_LEN);
185 0 : if (memcpy_s(ptr->memDesc, HCCL_MEM_DESC_STR_LEN, desc, descLen + 1) != EOK) {
186 0 : HCCL_ERROR("[HcclOneSidedService][RegMem] memcpy_s memDesc failed");
187 0 : return HCCL_E_INTERNAL;
188 : }
189 :
190 0 : if (firstReg) {
191 0 : registedMemCnt_++;
192 0 : std::string descStr(ptr->memDesc, HCCL_MEM_DESC_STR_LEN);
193 0 : if (useRdma) {
194 0 : desc2HcclBufMapRoce_.emplace(descStr, buf);
195 : } else {
196 0 : desc2HcclBufMapIpc_.emplace(descStr, buf);
197 : }
198 0 : }
199 0 : HCCL_DEBUG(
200 : "[HcclOneSidedService][RegMem] localRankId[%u] remoteRankId[%u] size[%lu] useRdma[%d] "
201 : "desc2HcclBufMap[%u] registedMemCnt[%u]",
202 : ptr->localRankId, ptr->remoteRankId, size, useRdma,
203 : useRdma ? desc2HcclBufMapRoce_.size() : desc2HcclBufMapIpc_.size(), registedMemCnt_);
204 0 : return HCCL_SUCCESS;
205 : }
206 :
207 0 : HcclBuf* HcclOneSidedService::GetHcclBufByDesc(std::string& descStr, bool useRdma)
208 : {
209 0 : HcclBuf* buf = nullptr;
210 0 : if (useRdma) {
211 0 : auto iter = desc2HcclBufMapRoce_.find(descStr);
212 0 : if (iter == desc2HcclBufMapRoce_.end()) {
213 0 : HCCL_ERROR("[HcclOneSidedService][GetHcclBufByDesc]Roce memory is not registered, please register first.");
214 0 : return nullptr;
215 : }
216 0 : buf = &(iter->second);
217 : } else {
218 0 : auto iter = desc2HcclBufMapIpc_.find(descStr);
219 0 : if (iter == desc2HcclBufMapIpc_.end()) {
220 0 : HCCL_ERROR("[HcclOneSidedService][GetHcclBufByDesc]Ipc memory is not registered, please register first.");
221 0 : return nullptr;
222 : }
223 0 : buf = &(iter->second);
224 : }
225 0 : return buf;
226 : }
227 :
228 0 : HcclResult HcclOneSidedService::DeregMem(const HcclMemDesc& localMemDesc)
229 : {
230 0 : const HcclMemDescData* ptr = static_cast<const HcclMemDescData*>(static_cast<const void*>(localMemDesc.desc));
231 0 : u32 remoteRankId = ptr->remoteRankId;
232 0 : if (registedMemCnt_ == 0) {
233 0 : HCCL_ERROR("[HcclOneSidedService][DeregMem]The number of registered memory is 0, please register first.");
234 0 : return HCCL_E_NOT_FOUND;
235 : }
236 :
237 0 : bool useRdma = true;
238 0 : if (isUsedRdmaMap_.find(remoteRankId) == isUsedRdmaMap_.end()) {
239 0 : CHK_RET(IsUsedRdma(remoteRankId, useRdma));
240 0 : isUsedRdmaMap_[remoteRankId] = useRdma;
241 : }
242 0 : useRdma = isUsedRdmaMap_[remoteRankId];
243 :
244 0 : std::string descStr(ptr->memDesc, HCCL_MEM_DESC_STR_LEN);
245 0 : HcclBuf* buf = GetHcclBufByDesc(descStr, useRdma);
246 0 : CHK_PRT_RET(
247 : buf == nullptr, HCCL_ERROR("[HcclOneSidedService][DeregMem] GetHcclBufByDesc failed."), HCCL_E_INTERNAL);
248 0 : HcclResult ret = HcclMemDereg(buf);
249 0 : if ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)) { // 调用DeregMem后，去注册的内存还需继续使用（即有多次注册
250 0 : return ret;
251 : }
252 :
253 0 : HCCL_DEBUG(
254 : "[HcclOneSidedService][DeregMem] localRankId[%u] remoteRankId[%u] size[%lu] useRdma[%d] "
255 : "desc2HcclBufMap[%u] registedMemCnt[%u]",
256 : ptr->localRankId, ptr->remoteRankId, buf->len, useRdma,
257 : useRdma ? desc2HcclBufMapRoce_.size() : desc2HcclBufMapIpc_.size(), registedMemCnt_);
258 :
259 0 : if (ret == HCCL_SUCCESS) {
260 0 : registedMemCnt_--;
261 0 : if (useRdma) {
262 0 : desc2HcclBufMapRoce_.erase(descStr);
263 : } else {
264 0 : desc2HcclBufMapIpc_.erase(descStr);
265 : }
266 : }
267 0 : return HCCL_SUCCESS;
268 0 : }
269 :
270 0 : HcclResult HcclOneSidedService::SetupRemoteRankInfo(RankId remoteRankId, HcclRankLinkInfo& remoteRankInfo)
271 : {
272 : // 检查 rankId 是否有效
273 0 : CHK_PRT_RET(
274 : rankTable_->rankList.size() <= remoteRankId,
275 : HCCL_ERROR(
276 : "[HcclOneSidedService][SetupRemoteRankInfo] the size of rankList is less than remoteRankId[%u].",
277 : remoteRankId),
278 : HCCL_E_NOT_FOUND);
279 :
280 0 : RankInfo_t tempRankInfo = rankTable_->rankList.at(remoteRankId);
281 0 : remoteRankInfo.userRank = tempRankInfo.rankId;
282 0 : remoteRankInfo.devicePhyId = tempRankInfo.deviceInfo.devicePhyId;
283 :
284 : // 检查 deviceIp 是否为空
285 0 : CHK_PRT_RET(
286 : tempRankInfo.deviceInfo.deviceIp.empty(),
287 : HCCL_ERROR("[HcclOneSidedService][SetupRemoteRankInfo] deviceIp is empty. RemoteRankId is [%u]", remoteRankId),
288 : HCCL_E_NOT_FOUND);
289 0 : remoteRankInfo.ip = tempRankInfo.deviceInfo.deviceIp[0];
290 :
291 0 : if (isUsedRdmaMap_.find(remoteRankId) != isUsedRdmaMap_.end() && !isUsedRdmaMap_[remoteRankId]) {
292 0 : bool useSuperPodMode = false;
293 0 : CHK_RET(IsSuperPodMode(useSuperPodMode));
294 :
295 0 : HcclIpAddress localVnicIp = HcclIpAddress(localRankInfo_.devicePhyId);
296 0 : HcclIpAddress remoteVnicIp = HcclIpAddress(remoteRankInfo.devicePhyId);
297 0 : RankInfo_t tRankInfo = rankTable_->rankList.at(localRankInfo_.userRank);
298 :
299 0 : if (useSuperPodMode) {
300 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
301 : localRankInfo_.devicePhyId, DeviceIdType::DEVICE_ID_TYPE_SDID, tRankInfo.superDeviceId, localVnicIp));
302 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
303 : localRankInfo_.devicePhyId, DeviceIdType::DEVICE_ID_TYPE_SDID, tempRankInfo.superDeviceId,
304 : remoteVnicIp));
305 : } else {
306 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
307 : localRankInfo_.devicePhyId, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, localRankInfo_.devicePhyId,
308 : localVnicIp));
309 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
310 : localRankInfo_.devicePhyId, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, remoteRankInfo.devicePhyId,
311 : remoteVnicIp));
312 : }
313 :
314 0 : localRankVnicInfo_.ip = localVnicIp;
315 0 : remoteRankInfo.ip = remoteVnicIp;
316 0 : }
317 0 : remoteRankInfo.port = tempRankInfo.deviceInfo.port == 0 || tempRankInfo.deviceInfo.port == HCCL_INVALID_PORT ?
318 : HETEROG_CCL_PORT :
319 0 : tempRankInfo.deviceInfo.port;
320 0 : remoteRankInfo.socketsPerLink = 1;
321 0 : return HCCL_SUCCESS;
322 0 : }
323 :
324 0 : HcclResult HcclOneSidedService::CreateLaunchStream()
325 : {
326 0 : g_launchStream = nullptr;
327 0 : constexpr u32 streamMode = 1; // 使能遇错即停
328 0 : EXCEPTION_CATCH(g_launchStream = std::make_unique<Stream>(StreamType::STREAM_TYPE_ONLINE), return HCCL_E_PTR);
329 0 : CHK_PTR_NULL(g_launchStream);
330 0 : CHK_PTR_NULL(g_launchStream->ptr());
331 0 : HCCL_INFO("[HcclOneSidedService][CreateLaunchStream] launchStream[%u]", g_launchStream->id());
332 0 : CHK_RET(hrtStreamSetMode(g_launchStream->ptr(), streamMode));
333 0 : return HCCL_SUCCESS;
334 : }
335 :
336 0 : HcclResult HcclOneSidedService::InitAicpuUnfoldMode()
337 : {
338 0 : if (isAicpuModeInited_) {
339 0 : return HCCL_SUCCESS;
340 : }
341 :
342 : DevType deviceType;
343 0 : CHK_RET(hrtGetDeviceType(deviceType));
344 0 : aicpuUnfoldMode_ = (deviceType == DevType::DEV_TYPE_910_93 || deviceType == DevType::DEV_TYPE_910B)
345 0 : && commConfig_.GetConfigAicpuUnfold(); // keep env flag for perf test
346 0 : HCCL_INFO(
347 : "[InitAicpuUnfoldMode] deviceType[%u] rdma[%u] aicpu[%u]", deviceType, (netDevRdmaCtx_ != nullptr),
348 : aicpuUnfoldMode_);
349 0 : if (aicpuUnfoldMode_) {
350 0 : CHK_PRT(LoadAICPUKernel());
351 0 : CHK_RET(AicpuResourceInit()); // 初始化service粒度资源
352 0 : CHK_RET(AicpuInitKernelLaunch());
353 : }
354 :
355 0 : isAicpuModeInited_ = true;
356 :
357 0 : return HCCL_SUCCESS;
358 : }
359 :
360 0 : HcclResult HcclOneSidedService::LoadAICPUKernel(void)
361 : {
362 0 : std::string jsonPath;
363 0 : CHK_RET(GetKernelFilePath(jsonPath));
364 0 : jsonPath += "ccl_kernel.json";
365 0 : HcclResult ret = LoadBinaryFromFile(jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0, binHandle_);
366 0 : CHK_PRT_RET(
367 : ret != HCCL_SUCCESS,
368 : HCCL_ERROR(
369 : "[LoadAICPUKernel]errNo[0x%016llx]load aicpu file fail, path[%s] optionType[%u]"
370 : "cpuKernelMode[%u].",
371 : ret, jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0),
372 : ret);
373 0 : return HCCL_SUCCESS;
374 0 : }
375 :
376 503 : void HcclOneSidedService::UnloadAICPUKernel(void)
377 : {
378 503 : if (binHandle_ != nullptr) {
379 0 : aclError aclRet = aclrtBinaryUnLoad(binHandle_);
380 0 : if (aclRet != ACL_SUCCESS) {
381 0 : HCCL_ERROR(
382 : "[UnloadAICPUKernel]errNo[0x%016llx] unload binary from binHandel[%p] error.", aclRet, binHandle_);
383 : }
384 0 : binHandle_ = nullptr;
385 : }
386 503 : return;
387 : }
388 :
389 0 : HcclResult HcclOneSidedService::CreateConnection(
390 : RankId remoteRankId, const HcclRankLinkInfo& remoteRankInfo, std::shared_ptr<HcclOneSidedConn>& tempConn)
391 : {
392 0 : CHK_RET(InitAicpuUnfoldMode());
393 0 : HcclNetDevCtx* ctx = isUsedRdmaMap_.at(remoteRankId) ? &netDevRdmaCtx_ : &netDevIpcCtx_;
394 0 : HcclRankLinkInfo* rankInfo = isUsedRdmaMap_.at(remoteRankId) ? &localRankInfo_ : &localRankVnicInfo_;
395 0 : u32 sdid = isUsedRdmaMap_.at(remoteRankId) ? 0 : rankTable_->rankList.at(localRankInfo_.userRank).superDeviceId;
396 0 : u32 serverId = isUsedRdmaMap_.at(remoteRankId) ? 0 : rankTable_->rankList.at(localRankInfo_.userRank).serverIdx;
397 : // 新增isNeedEnableP2P，用于判断remoteRank和本Rank是否在同一server上，在则需要enableP2P，反之则不需要
398 0 : bool isNeedEnableP2P = enableP2PRankIds_.find(remoteRankId) != enableP2PRankIds_.end();
399 0 : EXCEPTION_CATCH(
400 : tempConn = std::make_shared<HcclOneSidedConn>(
401 : *ctx, *rankInfo, remoteRankInfo, socketManager_, notifyPool_, dispatcher_, isUsedRdmaMap_[remoteRankId],
402 : sdid, serverId, trafficClass_, serviceLevel_, aicpuUnfoldMode_, isStandardCard_, isNeedEnableP2P),
403 : return HCCL_E_PTR);
404 0 : CHK_SMART_PTR_NULL(tempConn);
405 0 : return HCCL_SUCCESS;
406 : }
407 :
408 0 : HcclResult HcclOneSidedService::Grant(const HcclMemDesc& localMemDesc, const ProcessInfo& remoteProcess)
409 : {
410 0 : const HcclMemDescData* ptr = static_cast<const HcclMemDescData*>(static_cast<const void*>(localMemDesc.desc));
411 0 : std::string descStr(ptr->memDesc, HCCL_MEM_DESC_STR_LEN);
412 0 : HCCL_DEBUG("[HcclOneSidedService][Grant] desc[%s] length[%u]", descStr.c_str(), descStr.length());
413 0 : HcclBuf* buf = GetHcclBufByDesc(descStr, false);
414 0 : if (buf == nullptr) {
415 0 : return HCCL_E_INTERNAL;
416 : }
417 :
418 0 : HcclMemGrantInfo grantInfo = {remoteProcess.sdid, static_cast<int32_t>(remoteProcess.pid)};
419 0 : HcclResult ret = HcclMemGrant(buf, &grantInfo);
420 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[HcclOneSidedService][Grant] Grant error"), ret);
421 0 : return HCCL_SUCCESS;
422 0 : }
423 :
424 51 : HcclResult HcclOneSidedService::ExchangeMemDesc(
425 : RankId remoteRankId, const HcclMemDescs& localMemDescs, HcclMemDescs& remoteMemDescs, u32& actualNumOfRemote,
426 : const std::string& commIdentifier, s32 timeoutSec)
427 : {
428 51 : std::shared_ptr<HcclOneSidedConn> tempConn;
429 51 : std::unique_lock oneSidedConnslock(oneSidedConnsMutex_);
430 51 : auto it = oneSidedConns_.find(remoteRankId);
431 51 : if (it == oneSidedConns_.end()) {
432 51 : HcclRankLinkInfo remoteRankInfo;
433 51 : CHK_RET(SetupRemoteRankInfo(remoteRankId, remoteRankInfo));
434 0 : CHK_RET(CreateConnection(remoteRankId, remoteRankInfo, tempConn));
435 0 : timeoutSec = timeoutSec == 0 ? GetExternalInputHcclLinkTimeOut() : timeoutSec;
436 0 : CHK_RET(tempConn->Connect(commIdentifier, timeoutSec));
437 0 : oneSidedConns_.emplace(remoteRankId, tempConn);
438 51 : } else {
439 0 : tempConn = it->second;
440 : }
441 0 : std::unique_lock<std::mutex> lock(descMtx_);
442 0 : for (u32 i = 0; i < localMemDescs.arrayLength; ++i) {
443 0 : localMemDescs_[remoteRankId].push_back(localMemDescs.array[i]);
444 : }
445 0 : lock.unlock();
446 :
447 0 : return tempConn->ExchangeMemDesc(localMemDescs, remoteMemDescs, actualNumOfRemote);
448 51 : }
449 :
450 1 : void HcclOneSidedService::EnableMemAccess(const HcclMemDesc& remoteMemDesc, HcclMem& remoteMem)
451 : {
452 1 : HcclResult ret = HCCL_SUCCESS;
453 1 : const TransportMem::RmaMemDesc* ptr = reinterpret_cast<const TransportMem::RmaMemDesc*>(remoteMemDesc.desc);
454 1 : u32 remoteRank = ptr->localRankId;
455 1 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
456 1 : auto it = oneSidedConns_.find(remoteRank);
457 1 : if (it == oneSidedConns_.end()) {
458 1 : HCCL_ERROR(
459 : "[HcclOneSidedService][EnableMemAccess]connection not found, remoteRank[%u], "
460 : "please exchange mem desc to create connection first.",
461 : remoteRank);
462 1 : throw logic_error("[HcclOneSidedService][EnableMemAccess]connection not found.");
463 : }
464 0 : std::unique_lock<std::mutex> lock(descMtx_);
465 0 : auto descIt = localMemDescs_.find(remoteRank);
466 : // HCCS下进行权限授予
467 0 : if (!isUsedRdmaMap_[remoteRank] && descIt != localMemDescs_.end()) {
468 : s32 pid;
469 0 : SalGetBareTgid(&pid);
470 0 : RankId localRankId = localRankInfo_.userRank;
471 0 : u32 sid = rankTable_->rankList.at(localRankId).superDeviceId;
472 0 : u32 serverId = rankTable_->rankList.at(localRankId).serverIdx;
473 :
474 : // 收发进程信息
475 0 : ProcessInfo localProcess = {pid, sid, serverId};
476 0 : ProcessInfo remoteProcess = {};
477 :
478 0 : ret = it->second->ExchangeIpcProcessInfo(localProcess, remoteProcess);
479 0 : if (ret != HCCL_SUCCESS) {
480 0 : HCCL_ERROR(
481 : "[HcclOneSidedService][EnableMemAccess] Exchange ipc processInfo failed, ret[%d], "
482 : "remoteRank[%u].",
483 : ret, remoteRank);
484 0 : throw logic_error("[HcclOneSidedService][EnableMemAccess] Exchange ipc processInfo failed.");
485 : }
486 0 : remoteProcess.sdid = localProcess.serverId == remoteProcess.serverId ? INVALID_INT : remoteProcess.sdid;
487 :
488 0 : for (u32 i = 0; i < descIt->second.size(); ++i) {
489 0 : ret = Grant(descIt->second.at(i), remoteProcess);
490 0 : if (ret != HCCL_SUCCESS) {
491 0 : HCCL_ERROR(
492 : "[HcclOneSidedService][EnableMemAccess] Grant remote process failed, ret[%d], "
493 : "remoteRank[%u].",
494 : ret, remoteRank);
495 0 : throw logic_error("[HcclOneSidedService][EnableMemAccess] Grant remote process failed.");
496 : }
497 : }
498 0 : localMemDescs_.erase(descIt);
499 : }
500 0 : lock.unlock();
501 0 : it->second->EnableMemAccess(remoteMemDesc, remoteMem);
502 1 : }
503 :
504 1 : void HcclOneSidedService::DisableMemAccess(const HcclMemDesc& remoteMemDesc)
505 : {
506 1 : const TransportMem::RmaMemDesc* ptr = reinterpret_cast<const TransportMem::RmaMemDesc*>(remoteMemDesc.desc);
507 1 : u32 remoteRank = ptr->localRankId;
508 1 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
509 1 : if (oneSidedConns_.find(remoteRank) == oneSidedConns_.end()) {
510 1 : HCCL_ERROR(
511 : "[HcclOneSidedService][DisableMemAccess]connection not found by remoteRankId[%u], "
512 : "please exchange mem desc to create connection first.",
513 : remoteRank);
514 1 : throw logic_error("[HcclOneSidedService][DisableMemAccess]connection not found.");
515 : }
516 0 : oneSidedConns_.at(remoteRank)->DisableMemAccess(remoteMemDesc);
517 1 : }
518 :
519 400 : void HcclOneSidedService::BatchPut(
520 : RankId remoteRankId, const HcclOneSideOpDesc* desc, u32 descNum, const rtStream_t& stream)
521 : {
522 400 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
523 401 : auto it = oneSidedConns_.find(remoteRankId);
524 399 : if (it == oneSidedConns_.end()) {
525 400 : HCCL_ERROR("[HcclMemCommunication][BatchPut] Can't find oneSidedConn by remoteRank %u", remoteRankId);
526 401 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
527 : }
528 0 : if (aicpuUnfoldMode_) {
529 0 : EXCEPTION_THROW_IF_ERR(
530 : OrchestrateAicpu(remoteRankId, HcclCMDType::HCCL_CMD_BATCH_PUT, it->second, desc, descNum, stream),
531 : "[BatchPut] AICPU launch failed");
532 : } else {
533 0 : it->second->BatchWrite(desc, descNum, stream);
534 : }
535 401 : }
536 :
537 1 : void HcclOneSidedService::BatchGet(
538 : RankId remoteRankId, const HcclOneSideOpDesc* desc, u32 descNum, const rtStream_t& stream)
539 : {
540 1 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
541 1 : auto it = oneSidedConns_.find(remoteRankId);
542 1 : if (it == oneSidedConns_.end()) {
543 1 : HCCL_ERROR("[HcclMemCommunication][BatchGet] Can't find oneSidedConn by remoteRank %u", remoteRankId);
544 1 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
545 : }
546 0 : if (aicpuUnfoldMode_) {
547 0 : EXCEPTION_THROW_IF_ERR(
548 : OrchestrateAicpu(remoteRankId, HcclCMDType::HCCL_CMD_BATCH_GET, it->second, desc, descNum, stream),
549 : "[BatchGet] AICPU launch failed");
550 : } else {
551 0 : it->second->BatchRead(desc, descNum, stream);
552 : }
553 1 : }
554 :
555 : // 绑定一块全局内存
556 0 : HcclResult HcclOneSidedService::BindMem(void* memRecordHandle, const std::string& commIdentifier)
557 : {
558 0 : auto memRecordPtr = static_cast<GlobalMemRecord*>(memRecordHandle);
559 0 : CHK_RET(memRecordPtr->BindToComm(commIdentifier));
560 :
561 : // 是否重复绑定在前面BindToComm已经检查过了
562 0 : auto emplaceResult = boundMemPtrSet_.emplace(memRecordPtr);
563 0 : CHK_PRT_RET(
564 : emplaceResult.second == false,
565 : HCCL_ERROR(
566 : "[HcclOneSidedService][BindMem] Emplace mem record ptr failed, memRecordPtr[%p], comm[%s].", memRecordPtr,
567 : commIdentifier.c_str()),
568 : HCCL_E_INTERNAL);
569 :
570 0 : HCCL_INFO(
571 : "[HcclOneSidedService][BindMem] Bind mem successfully, memHandle[%p], comm[%s].", memRecordHandle,
572 : commIdentifier.c_str());
573 0 : return HCCL_SUCCESS;
574 : }
575 :
576 : // 解绑一块全局内存
577 0 : HcclResult HcclOneSidedService::UnbindMem(void* memRecordHandle, const std::string& commIdentifier)
578 : {
579 0 : auto memRecordPtr = static_cast<GlobalMemRecord*>(memRecordHandle);
580 0 : CHK_RET(memRecordPtr->UnbindFromComm(commIdentifier));
581 :
582 0 : const auto eraseCount = boundMemPtrSet_.erase(memRecordPtr);
583 0 : CHK_PRT_RET(
584 : eraseCount == 0,
585 : HCCL_ERROR(
586 : "[HcclOneSidedService][UnbindMem] Erase mem record ptr failed, memRecordPtr[%p], comm[%s].",
587 : memRecordHandle, commIdentifier.c_str()),
588 : HCCL_E_INTERNAL);
589 :
590 0 : HCCL_INFO(
591 : "[HcclOneSidedService][UnbindMem] Unbind mem successfully, memHandle[%p], comm[%s].", memRecordHandle,
592 : commIdentifier.c_str());
593 0 : return HCCL_SUCCESS;
594 : }
595 :
596 0 : HcclResult HcclOneSidedService::DeInit()
597 : {
598 0 : if (aicpuUnfoldMode_) {
599 0 : std::unique_lock<std::mutex> guard{g_launchMutex};
600 0 : CHK_RET(CreateLaunchStream());
601 0 : CHK_RET(OrchestrateAicpu(0, HcclCMDType::HCCL_CMD_BATCH_GET, nullptr, nullptr, 0, g_launchStream->ptr()));
602 0 : CHK_RET(hcclStreamSynchronize(
603 : g_launchStream->ptr(), CommConfiger::GetInstance().GetCommConfigExecTimeOut(identifier_)));
604 0 : HCCL_INFO("[HcclOneSidedService][DeInit] destroy launchStream[%u]", g_launchStream->id());
605 0 : g_launchStream = nullptr;
606 0 : }
607 :
608 : // 检查是否还绑定着全局内存
609 0 : if (!boundMemPtrSet_.empty()) {
610 0 : HCCL_ERROR("[HcclOneSidedService][DeInit] There are memories still bound to this comm; please unbind them "
611 : "before destroying the comm.");
612 0 : HCCL_ERROR("[HcclOneSidedService][DeInit] List of bound memories:");
613 0 : for (auto handle : boundMemPtrSet_) {
614 0 : auto memRecordPtr = static_cast<GlobalMemRecord*>(handle);
615 0 : const auto info = memRecordPtr->PrintInfo();
616 0 : HCCL_ERROR("[HcclOneSidedService][DeInit][Bound mem] ptr:%p, %s", handle, info.c_str());
617 0 : }
618 0 : return HCCL_E_PARA;
619 : }
620 :
621 0 : if (prepared_) {
622 : // 去使能内存
623 0 : CHK_RET(DisableMemAccess());
624 0 : prepared_ = false;
625 : }
626 0 : UnloadAICPUKernel();
627 0 : return HCCL_SUCCESS;
628 : }
629 :
630 0 : HcclResult HcclOneSidedService::RunFuncWithTimeout(
631 : std::function<HcclResult()> func, const std::string& commIdentifier, s32 timeoutSec, std::string functionName)
632 : {
633 0 : std::future<HcclResult> futureResult;
634 0 : futureResult = std::async(std::launch::async, func);
635 :
636 0 : CHK_PRT_RET(
637 : !futureResult.valid(),
638 : HCCL_ERROR("[HcclOneSidedService][%s] futureResult is not assigned.", functionName.c_str()), HCCL_E_INTERNAL);
639 :
640 : // 超时检查，若timeout设置为-1则不检查，上层已经保证timeout不会为0
641 0 : if (timeoutSec != -1 && futureResult.wait_for(std::chrono::seconds(timeoutSec)) == std::future_status::timeout) {
642 : // 发生超时，设置stop flag让socket线程停止，避免进程长时间无法退出
643 0 : CHK_RET(socketManager_->SetStopFlag(true));
644 0 : HCCL_ERROR(
645 : "[HcclOneSidedService][%s]timeout. commIdentifier[%s], timeout[%ds]", functionName.c_str(),
646 : commIdentifier.c_str(), timeoutSec);
647 0 : futureResult.wait();
648 0 : CHK_RET(socketManager_->SetStopFlag(false));
649 0 : return HCCL_E_TIMEOUT;
650 : }
651 :
652 0 : HcclResult ret = futureResult.get();
653 0 : CHK_PRT_RET(
654 : ret != HCCL_SUCCESS,
655 : HCCL_ERROR(
656 : "[HcclOneSidedService][%s] Prepare failed. commIdentifier[%s]", functionName.c_str(),
657 : commIdentifier.c_str()),
658 : ret);
659 :
660 0 : return HCCL_SUCCESS;
661 0 : }
662 :
663 1 : HcclResult HcclOneSidedService::PrepareFullMesh(const std::string& commIdentifier, s32 timeoutSec)
664 : {
665 1 : HcclUs startut = TIME_NOW();
666 : // 创建连接
667 1 : CHK_RET(CreateLinkFullmesh(commIdentifier, timeoutSec));
668 : // 注册内存
669 0 : CHK_RET(RegisterBoundMems());
670 : // 交换内存描述符
671 0 : CHK_RET(RunFuncWithTimeout(
672 : [this]() -> HcclResult {
673 : return this->ExchangeMemDescFullMesh();
674 : },
675 : commIdentifier, timeoutSec, "ExchangeMemDescFullMesh"));
676 : // 使能访问
677 0 : CHK_RET(RunFuncWithTimeout(
678 : [this]() -> HcclResult {
679 : return this->EnableMemAccessByThread();
680 : },
681 : commIdentifier, timeoutSec, "EnableMemAccessByThread"));
682 :
683 0 : HCCL_INFO(
684 : "[HcclOneSidedService][PrepareFullMesh] Prepare finished. comm[%s], take time [%lld us].",
685 : commIdentifier.c_str(), DURATION_US(TIME_NOW() - startut));
686 0 : return HCCL_SUCCESS;
687 : }
688 :
689 : HcclResult
690 5 : HcclOneSidedService::Prepare(const std::string& commIdentifier, const HcclPrepareConfig* prepareConfig, s32 timeoutSec)
691 : {
692 : // 如果已经prepare过，直接返回Success
693 5 : CHK_PRT_RET(
694 : prepared_, HCCL_WARNING("[HcclOneSidedService][Prepare] This comm[%s] has prepared.", commIdentifier.c_str()),
695 : HCCL_SUCCESS);
696 :
697 4 : CHK_RET(hrtGetDevice(&deviceLogicId_));
698 :
699 4 : if (needRegIpcMem_) {
700 0 : SalGetBareTgid(&localProcess_.pid);
701 0 : RankId localRankId = localRankInfo_.userRank;
702 0 : localProcess_.sdid = rankTable_->rankList.at(localRankId).superDeviceId;
703 0 : localProcess_.serverId = rankTable_->rankList.at(localRankId).serverIdx;
704 : }
705 :
706 4 : HcclTopoType configTopoType = prepareConfig->topoType;
707 4 : std::future<HcclResult> futureResult;
708 4 : timeoutSec = timeoutSec == 0 ? GetExternalInputHcclLinkTimeOut() : timeoutSec;
709 4 : if (configTopoType == HcclTopoType::HCCL_TOPO_FULLMESH) {
710 3 : HCCL_INFO("[HcclOneSidedService][Prepare] topoType is fullmesh.");
711 :
712 3 : auto ret = PrepareFullMesh(commIdentifier, timeoutSec);
713 3 : if (ret != HCCL_SUCCESS) {
714 2 : u32 rankSize = (rankTable_->rankList).size();
715 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
716 2 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
717 2 : auto it = oneSidedConns_.find(remoteRankId);
718 2 : if (it == oneSidedConns_.end()) {
719 : // remoteRankId超出oneSidedConns_的范围，直接退出
720 2 : HCCL_ERROR(
721 : "[HcclOneSidedService][Prepare] remoteRankId[%u] "
722 : "is not found in map oneSidedConns.",
723 : remoteRankId);
724 2 : break;
725 : }
726 0 : if (remoteRankId == localRankInfo_.userRank || oneSidedConns_.at(remoteRankId) == nullptr) {
727 0 : HCCL_INFO("[HcclOneSidedService][CleanSocketResource] remoteRank[%u] skip.", remoteRankId);
728 0 : continue;
729 : }
730 0 : HCCL_INFO("[HcclOneSidedService][CleanSocketResource] remote[%u]", remoteRankId);
731 0 : oneSidedConns_.at(remoteRankId)->CleanSocketResource(commIdentifier);
732 : }
733 2 : HCCL_ERROR("[HcclOneSidedService][Prepare] Prepare failed. commIdentifier[%s]", commIdentifier.c_str());
734 2 : return ret;
735 2 : }
736 : }
737 :
738 2 : prepared_ = true;
739 2 : return HCCL_SUCCESS;
740 4 : }
741 :
742 0 : HcclResult HcclOneSidedService::InitIsUsedRdmaMap(bool& needInitNic, bool& needInitVnic)
743 : {
744 0 : u32 rankSize = (rankTable_->rankList).size();
745 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
746 0 : if (remoteRankId == localRankInfo_.userRank) {
747 0 : continue;
748 : }
749 : bool isUseRdma;
750 0 : CHK_RET(IsUsedRdma(remoteRankId, isUseRdma));
751 0 : isUsedRdmaMap_[remoteRankId] = isUseRdma;
752 :
753 0 : if (isUseRdma) {
754 0 : needRegRoceMem_ = true;
755 : } else {
756 0 : needRegIpcMem_ = true;
757 : }
758 : }
759 0 : needInitNic = needRegRoceMem_;
760 0 : needInitVnic = needRegIpcMem_;
761 :
762 0 : HCCL_INFO(
763 : "[HcclOneSidedService][InitIsUsedRdmaMap] needInitNic is [%d], needInitVnic is [%d]", needInitNic,
764 : needInitVnic);
765 0 : return HCCL_SUCCESS;
766 : }
767 :
768 0 : void HcclOneSidedService::ConnectByThread(
769 : std::shared_ptr<HcclOneSidedConn>& conn, const std::string& commIdentifier, s32 timeoutSec, HcclResult& retOut)
770 : {
771 0 : if (deviceLogicId_ != HOST_DEVICE_ID) {
772 0 : hrtSetDevice(deviceLogicId_);
773 : }
774 0 : HcclResult ret = conn->ConnectWithRemote(commIdentifier, localProcess_, timeoutSec);
775 0 : retOut = ret;
776 0 : if (ret != HCCL_SUCCESS) {
777 0 : hasErrorFlag_ = true;
778 0 : if (ret == HCCL_E_TIMEOUT) {
779 0 : hasTimeoutErrorFlag_ = true;
780 : }
781 0 : HCCL_ERROR("[ConnectByThread] Connect failed. userrank[%u], ret[%d].", localRankInfo_.userRank, ret);
782 : }
783 0 : hrtResetDevice(deviceLogicId_);
784 0 : }
785 :
786 0 : HcclResult HcclOneSidedService::CreateLinkFullmesh(const std::string& commIdentifier, s32 timeoutSec)
787 : {
788 0 : u32 rankSize = (rankTable_->rankList).size();
789 :
790 0 : std::unique_lock oneSidedConnslock(oneSidedConnsMutex_);
791 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
792 0 : if (remoteRankId == localRankInfo_.userRank) {
793 0 : continue;
794 : }
795 0 : HcclRankLinkInfo remoteRankInfo;
796 0 : CHK_RET(SetupRemoteRankInfo(remoteRankId, remoteRankInfo));
797 0 : CHK_RET(CreateConnection(remoteRankId, remoteRankInfo, oneSidedConns_[remoteRankId]));
798 0 : }
799 :
800 0 : std::vector<std::unique_ptr<std::thread>> linkThreads;
801 0 : std::vector<HcclResult> linkResult;
802 0 : linkThreads.resize(rankSize);
803 0 : linkResult.resize(rankSize, HCCL_SUCCESS);
804 0 : hasErrorFlag_ = false;
805 0 : ThreadsGuard threadsGuard(linkThreads);
806 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
807 0 : if (remoteRankId == localRankInfo_.userRank) {
808 0 : continue;
809 : }
810 0 : linkThreads[remoteRankId].reset(new (std::nothrow) std::thread(
811 0 : &HcclOneSidedService::ConnectByThread, this, std::ref(oneSidedConns_.at(remoteRankId)), commIdentifier,
812 0 : timeoutSec, std::ref(linkResult[remoteRankId])));
813 0 : CHK_SMART_PTR_NULL(linkThreads[remoteRankId]);
814 : }
815 0 : oneSidedConnslock.unlock();
816 :
817 0 : for (u32 remoteRankId = 0; remoteRankId < linkThreads.size(); remoteRankId++) {
818 0 : if (linkThreads[remoteRankId] == nullptr || !linkThreads[remoteRankId]->joinable()) {
819 0 : continue;
820 : }
821 0 : linkThreads[remoteRankId]->join(); // 等待线程执行完毕
822 : }
823 0 : linkThreads.clear();
824 :
825 0 : for (u32 remoteRankId = 0; remoteRankId < linkResult.size(); remoteRankId++) {
826 0 : CHK_PRT_RET(
827 : linkResult[remoteRankId] != HCCL_SUCCESS,
828 : HCCL_ERROR(
829 : "[HcclOneSidedService][CreateLinkFullmesh] Create links failed. commIdentifier[%s].",
830 : commIdentifier.c_str()),
831 : linkResult[remoteRankId]);
832 : }
833 :
834 0 : CHK_PRT_RET(
835 : hasErrorFlag_ == true,
836 : HCCL_ERROR(
837 : "[HcclOneSidedService][CreateLinkFullmesh] Create links failed. commIdentifier[%s].",
838 : commIdentifier.c_str()),
839 : hasTimeoutErrorFlag_ ? HCCL_E_TIMEOUT : HCCL_E_INTERNAL);
840 :
841 0 : HCCL_INFO(
842 : "[HcclOneSidedService][CreateLinkFullmesh] Create links success. commIdentifier[%s].", commIdentifier.c_str());
843 0 : return HCCL_SUCCESS;
844 0 : }
845 :
846 0 : HcclResult HcclOneSidedService::RegBoundMem(
847 : HcclNetDevCtx netDevCtx, const HcclMem& localMem, HcclMemDesc& localMemDesc, HcclBuf& buf)
848 : {
849 0 : std::unique_lock<std::mutex> lock(regMutex_);
850 0 : HcclResult ret = HcclMemReg(netDevCtx, &localMem, &buf);
851 0 : if ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)) { // HCCL_E_AGAIN:调用HcclMemReg前，内存已注册过
852 0 : return ret;
853 : }
854 :
855 0 : char* desc = nullptr;
856 0 : uint64_t descLen = 0;
857 0 : ret = HcclMemExport(&buf, &desc, &descLen);
858 0 : if (ret != HCCL_SUCCESS) {
859 0 : HCCL_ERROR("[HcclOneSidedService][RegBoundMem] get mem desc failed, ret[%d]", ret);
860 0 : throw logic_error("[HcclOneSidedService][RegBoundMem] get mem desc failed");
861 : }
862 0 : lock.unlock();
863 :
864 0 : HcclMemDescData* ptr = static_cast<HcclMemDescData*>(static_cast<void*>(localMemDesc.desc));
865 0 : ptr->localRankId = localRankInfo_.userRank;
866 0 : ptr->remoteRankId = INVALID_REMOTE_RANK_ID; // 进程粒度注册，不区分对端rank, 填为全F
867 0 : memset_s(ptr->memDesc, HCCL_MEM_DESC_STR_LEN, 0, HCCL_MEM_DESC_STR_LEN);
868 0 : if (memcpy_s(ptr->memDesc, HCCL_MEM_DESC_STR_LEN, desc, descLen + 1) != EOK) {
869 0 : HCCL_ERROR("[HcclOneSidedService][RegBoundMem] memcpy_s memDesc failed");
870 0 : return HCCL_E_INTERNAL;
871 : }
872 :
873 0 : HCCL_INFO("[HcclOneSidedService][RegBoundMem] RegBoundMem success. addr[%p], size[%llu].", buf.addr, buf.len);
874 0 : return HCCL_SUCCESS;
875 0 : }
876 :
877 0 : HcclResult HcclOneSidedService::RegisterBoundMems()
878 : {
879 0 : localMemIpcDescs_.reserve(boundMemPtrSet_.size());
880 0 : localMemRoceDescs_.reserve(boundMemPtrSet_.size());
881 0 : localMemIpcDescs_.clear();
882 0 : localMemRoceDescs_.clear();
883 0 : for (auto& recordPtr : boundMemPtrSet_) {
884 0 : HcclMem mem{recordPtr->GetMemType(), const_cast<void*>(recordPtr->GetAddr()), recordPtr->GetSize()};
885 0 : if (needRegRoceMem_) {
886 : HcclBuf buf;
887 : HcclMemDesc localMemDesc;
888 0 : CHK_RET(RegBoundMem(netDevRdmaCtx_, mem, localMemDesc, buf));
889 0 : localMemRoceDescs_.push_back(localMemDesc);
890 0 : recordPtr->SaveRegBufInfo(netDevRdmaCtx_, buf);
891 : }
892 0 : if (needRegIpcMem_) {
893 : HcclBuf buf;
894 : HcclMemDesc localMemDesc;
895 0 : CHK_RET(RegBoundMem(netDevIpcCtx_, mem, localMemDesc, buf));
896 0 : if (recordPtr->GetMemType() == HCCL_MEM_TYPE_DEVICE) {
897 0 : localMemIpcDescs_.push_back(localMemDesc);
898 : }
899 0 : recordPtr->SaveRegBufInfo(netDevIpcCtx_, buf);
900 0 : CHK_RET(Grant(buf));
901 : }
902 : }
903 0 : HCCL_INFO("[HcclOneSidedService][RegisterBoundMems] Register bound mems success.");
904 0 : return HCCL_SUCCESS;
905 : }
906 :
907 0 : HcclResult HcclOneSidedService::ExchangeMemDescFullMesh()
908 : {
909 0 : u32 rankSize = (rankTable_->rankList).size();
910 0 : std::vector<std::unique_ptr<std::thread>> exchangeThreads;
911 0 : exchangeThreads.resize(rankSize);
912 :
913 0 : hasErrorFlag_ = false;
914 0 : ThreadsGuard threadsGuard(exchangeThreads);
915 0 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
916 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
917 0 : if (remoteRankId == localRankInfo_.userRank) {
918 0 : continue;
919 : }
920 0 : exchangeThreads[remoteRankId].reset(new (std::nothrow) std::thread(
921 0 : &HcclOneSidedService::ExchangeMemDescByThread, this, std::ref(oneSidedConns_.at(remoteRankId)),
922 0 : isUsedRdmaMap_[remoteRankId]));
923 0 : CHK_SMART_PTR_NULL(exchangeThreads[remoteRankId]);
924 : }
925 0 : oneSidedConnslock.unlock();
926 :
927 0 : for (u32 remoteRankId = 0; remoteRankId < exchangeThreads.size(); remoteRankId++) {
928 0 : if (exchangeThreads[remoteRankId] == nullptr || !exchangeThreads[remoteRankId]->joinable()) {
929 0 : continue;
930 : }
931 0 : exchangeThreads[remoteRankId]->join(); // 等待线程执行完毕
932 : }
933 0 : CHK_PRT_RET(
934 : hasErrorFlag_ == true, HCCL_ERROR("[HcclOneSidedService][ExchangeMemDescFullMesh] Exchange mem desc failed."),
935 : HCCL_E_INTERNAL);
936 :
937 0 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDescFullMesh] Exchange mem desc success.");
938 0 : return HCCL_SUCCESS;
939 0 : }
940 :
941 0 : HcclResult HcclOneSidedService::ExchangeMemDescByThread(std::shared_ptr<HcclOneSidedConn>& conn, bool isUseRdma)
942 : {
943 0 : if (deviceLogicId_ != HOST_DEVICE_ID) {
944 0 : hrtSetDevice(deviceLogicId_);
945 : }
946 :
947 : HcclMemDescs localMemDescs;
948 0 : if (isUseRdma) {
949 0 : localMemDescs.array = localMemRoceDescs_.data();
950 0 : localMemDescs.arrayLength = localMemRoceDescs_.size();
951 : } else {
952 0 : localMemDescs.array = localMemIpcDescs_.data();
953 0 : localMemDescs.arrayLength = localMemIpcDescs_.size();
954 : }
955 :
956 0 : HcclResult ret = conn->ExchangeMemDesc(localMemDescs);
957 0 : if (ret != HCCL_SUCCESS) {
958 0 : hasErrorFlag_ = true;
959 0 : HCCL_ERROR(
960 : "[ExchangeMemDescByThread] ExchangeMemDescByThread failed. userRank[%u], ret[%d].", localRankInfo_.userRank,
961 : ret);
962 : }
963 0 : CHK_RET(hrtResetDevice(deviceLogicId_));
964 0 : return HCCL_SUCCESS;
965 : }
966 :
967 0 : HcclResult HcclOneSidedService::EnableMemAccessByThread()
968 : {
969 0 : if (deviceLogicId_ != HOST_DEVICE_ID) {
970 0 : hrtSetDevice(deviceLogicId_);
971 : }
972 0 : CHK_RET(EnableMemAccess());
973 0 : CHK_RET(hrtResetDevice(deviceLogicId_));
974 0 : return HCCL_SUCCESS;
975 : }
976 :
977 0 : HcclResult HcclOneSidedService::EnableMemAccess()
978 : {
979 0 : u32 rankSize = (rankTable_->rankList).size();
980 0 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
981 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
982 0 : if (remoteRankId == localRankInfo_.userRank) {
983 0 : continue;
984 : }
985 0 : CHK_RET(oneSidedConns_.at(remoteRankId)->EnableMemAccess());
986 : }
987 0 : return HCCL_SUCCESS;
988 0 : }
989 :
990 0 : HcclResult HcclOneSidedService::DisableMemAccess()
991 : {
992 0 : u32 rankSize = (rankTable_->rankList).size();
993 0 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
994 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
995 0 : if (remoteRankId == localRankInfo_.userRank) {
996 0 : continue;
997 : }
998 0 : CHK_RET(oneSidedConns_.at(remoteRankId)->DisableMemAccess());
999 : }
1000 0 : return HCCL_SUCCESS;
1001 0 : }
1002 :
1003 0 : HcclResult HcclOneSidedService::Grant(HcclBuf& buf)
1004 : {
1005 0 : u32 rankSize = (rankTable_->rankList).size();
1006 0 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
1007 0 : for (u32 remoteRankId = 0; remoteRankId < rankSize; remoteRankId++) {
1008 0 : if (remoteRankId == localRankInfo_.userRank || isUsedRdmaMap_[remoteRankId] == true) {
1009 0 : continue;
1010 : }
1011 : ProcessInfo remoteProcess;
1012 0 : CHK_RET(oneSidedConns_.at(remoteRankId)->GetRemoteProcessInfo(remoteProcess));
1013 0 : HcclMemGrantInfo grantInfo = {remoteProcess.sdid, static_cast<int32_t>(remoteProcess.pid)};
1014 :
1015 0 : HcclResult ret = HcclMemGrant(&buf, &grantInfo);
1016 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[HcclOneSidedService][Grant] Grant error"), ret);
1017 : }
1018 0 : return HCCL_SUCCESS;
1019 0 : }
1020 :
1021 0 : HcclResult HcclOneSidedService::OrchestrateAicpu(
1022 : RankId remoteRankId, HcclCMDType cmdType, const std::shared_ptr<HcclOneSidedConn>& conn,
1023 : const HcclOneSideOpDesc* desc, u32 descNum, rtStream_t stream)
1024 : {
1025 : bool useRdma;
1026 0 : CHK_RET(GetIsUsedRdma(remoteRankId, useRdma));
1027 :
1028 0 : HCCL_DEBUG(
1029 : "[OrchestrateAicpu] aicpu unfold launch kernel: desc[%p] descNum[%u] cmdType[%u] tag[%s] localRank[%u] "
1030 : "remoteRank[%u] useRdma[%d]",
1031 : desc, descNum, cmdType, identifier_.c_str(), localRankInfo_.userRank, remoteRankId, useRdma);
1032 :
1033 0 : AicpuOneSideCommTiling tilingInfo;
1034 0 : tilingInfo.cmdType = cmdType;
1035 0 : tilingInfo.tag = identifier_;
1036 0 : tilingInfo.stream = stream;
1037 0 : tilingInfo.dumpDebug = GetExternalInputHcclDumpDebug();
1038 0 : tilingInfo.useRdma = useRdma;
1039 0 : aclrtFloatOverflowMode floatOverflowMode = ACL_RT_OVERFLOW_MODE_UNDEF;
1040 0 : CHK_RET(hrtGetDeviceSatMode(&floatOverflowMode));
1041 0 : tilingInfo.floatOverflowMode = floatOverflowMode;
1042 0 : const u64 dynamicDataSize = CalcTilingDynamicDataSize(cmdType, descNum);
1043 0 : CHK_RET(InitAicpuTilingDataBuf(tilingInfo, remoteRankId, conn, desc, descNum, dynamicDataSize));
1044 : // 根据算子类型，获取 Aicpu Kernel 名称
1045 0 : auto iter = HCOM_CMD_TYPE_STR_MAP.find(cmdType);
1046 0 : CHK_PRT_RET(
1047 : (iter == HCOM_CMD_TYPE_STR_MAP.end()),
1048 : HCCL_ERROR("[%s] RunAicpuRpcSrvLaunchV2 kernel not found, cmdType=[%d]", __func__, static_cast<int>(cmdType)),
1049 : HCCL_E_INTERNAL);
1050 0 : std::string kernelName = std::string("RunAicpuRpcSrvLaunchV2") + "_" + iter->second;
1051 0 : HcclResult ret = AicpuKernelLaunch(conn, kernelName, tilingInfo, sizeof(struct OpTilingData) + dynamicDataSize);
1052 0 : CHK_PRT_RET(
1053 : ret != HCCL_SUCCESS,
1054 : HCCL_ERROR(
1055 : "[OrchestrateAicpu] aicpu unfold launch kernel[%s] failed. ret[%u], "
1056 : "desc[%p] descNum[%u] cmdType[%u] tag[%s]",
1057 : kernelName.c_str(), ret, desc, descNum, cmdType, identifier_.c_str()),
1058 : ret);
1059 0 : return HCCL_SUCCESS;
1060 0 : }
1061 :
1062 0 : HcclResult HcclOneSidedService::AicpuResourceInit()
1063 : {
1064 0 : const u64 beginTime = hrtMsprofSysCycleTime();
1065 :
1066 0 : execStream_ = Stream(StreamType::STREAM_TYPE_DEVICE);
1067 0 : const u32 streamContextSize = sizeof(SqCqeContext);
1068 0 : commResPara_.execStreamParam.streamInfo.streamIds = execStream_.id();
1069 0 : commResPara_.execStreamParam.streamInfo.sqIds = execStream_.sqId();
1070 0 : commResPara_.execStreamParam.streamInfo.cqIds = execStream_.cqId();
1071 0 : commResPara_.execStreamParam.streamInfo.logicCqids = execStream_.logicCqId();
1072 0 : CHK_RET(DeviceMem::alloc(execStreamContext_, streamContextSize));
1073 0 : CHK_RET(hrtMemSet(execStreamContext_.ptr(), streamContextSize, streamContextSize));
1074 0 : commResPara_.execStreamParam.sqCqContextAddr = reinterpret_cast<u64>(execStreamContext_.ptr());
1075 0 : commResPara_.execStreamParam.sqCqContextSize = streamContextSize;
1076 :
1077 0 : const u32 postNotifyIdx = static_cast<u32>(AicpuLocalNotify::HOST_TO_AICPU_POST);
1078 0 : HcclResult ret = CreateAicpuNotify(localAicpuNotify_[postNotifyIdx], commResPara_.aicpuOpNotify[postNotifyIdx]);
1079 0 : CHK_PRT_RET(
1080 : ret != HCCL_SUCCESS,
1081 : HCCL_ERROR("[AicpuResourceInit] create aicpu post notify failed, errNo[0x%016llx]", HCCL_ERROR_CODE(ret)), ret);
1082 0 : const u32 waitNotifyIdx = static_cast<u32>(AicpuLocalNotify::HOST_TO_AICPU_WAIT);
1083 0 : ret = CreateAicpuNotify(localAicpuNotify_[waitNotifyIdx], commResPara_.aicpuOpNotify[waitNotifyIdx]);
1084 0 : CHK_PRT_RET(
1085 : ret != HCCL_SUCCESS,
1086 : HCCL_ERROR("[AicpuResourceInit] create aicpu wait notify failed, errNo[0x%016llx]", HCCL_ERROR_CODE(ret)), ret);
1087 :
1088 0 : CHK_RET(DeviceMem::alloc(commResParaDevice_, sizeof(HcclOneSideCommResParam)));
1089 :
1090 0 : const u64 endTime = hrtMsprofSysCycleTime();
1091 0 : HCCL_DEBUG("[AicpuResourceInit] done, time cost[%llu]", (endTime - beginTime));
1092 :
1093 0 : return HCCL_SUCCESS;
1094 : }
1095 :
1096 0 : HcclResult HcclOneSidedService::ReportProfilingCommInfo(const Stream& kfcStream, const Stream& aicpuStream)
1097 : {
1098 : ProfilingDeviceCommResInfo profCommInfo;
1099 0 : profCommInfo.groupName = hrtMsprofGetHashId(identifier_.c_str(), identifier_.length());
1100 0 : profCommInfo.rankSize = rankTable_->rankNum;
1101 0 : profCommInfo.rankId = localRankInfo_.userRank;
1102 0 : profCommInfo.usrRankId = localRankInfo_.userRank;
1103 0 : profCommInfo.aicpuKfcStreamId = static_cast<uint32_t>(kfcStream.id());
1104 0 : profCommInfo.reserve = 0;
1105 0 : HCCL_INFO(
1106 : "[ReportProfilingCommInfo] group[%s], groupHashId[%llu], streamId[%u]", identifier_.c_str(),
1107 : profCommInfo.groupName, aicpuStream.id());
1108 0 : profCommInfo.commStreamIds[0] = aicpuStream.id();
1109 0 : profCommInfo.commStreamSize = 1; // 只有1条执行流
1110 0 : return ProfilingManagerPub::CallMsprofReportMc2CommInfo(
1111 0 : hrtMsprofSysCycleTime(), &profCommInfo, sizeof(profCommInfo));
1112 : }
1113 :
1114 0 : HcclResult HcclOneSidedService::AicpuInitKernelLaunch()
1115 : {
1116 0 : const u64 beginTime = hrtMsprofSysCycleTime();
1117 :
1118 : {
1119 0 : std::unique_lock<std::mutex> guard{g_launchMutex};
1120 : struct InitTask {
1121 : u64 context; // A矩阵地址，通信在前时为sendbuffer
1122 : bool isCustom;
1123 : };
1124 0 : InitTask initTask = {};
1125 0 : initTask.context = 0ULL;
1126 0 : initTask.isCustom = false;
1127 0 : u16 timeOut = 0;
1128 0 : char kernelName[64] = "RunAicpuKfcResInitV2";
1129 0 : CHK_RET(CreateLaunchStream());
1130 0 : CHK_RET(AicpuAclKernelLaunch(
1131 : g_launchStream->ptr(), reinterpret_cast<void*>(&initTask), sizeof(initTask), binHandle_, kernelName, true,
1132 : timeOut));
1133 0 : CHK_RET(hcclStreamSynchronize(
1134 : g_launchStream->ptr(), CommConfiger::GetInstance().GetCommConfigExecTimeOut(identifier_)));
1135 0 : HCCL_RUN_INFO(
1136 : "[AicpuInitKernelLaunch] launch in launchStream[%u], execStream[%u]", g_launchStream->id(),
1137 : execStream_.id());
1138 0 : g_launchStream = nullptr;
1139 0 : }
1140 :
1141 0 : const u64 endTime = hrtMsprofSysCycleTime();
1142 0 : s32 threadId = SalGetTid();
1143 0 : std::string profName = "OneSideCommAicpuInit";
1144 0 : CHK_RET(ProfilingManagerPub::CallMsprofReportNodeInfo(beginTime, endTime, profName, threadId));
1145 :
1146 0 : return HCCL_SUCCESS;
1147 0 : }
1148 :
1149 0 : HcclResult HcclOneSidedService::CreateAicpuNotify(std::shared_ptr<LocalNotify>& localNotify, HcclSignalInfo& notifyInfo)
1150 : {
1151 0 : EXCEPTION_CATCH((localNotify = std::make_shared<LocalNotify>()), return HCCL_E_PTR);
1152 0 : CHK_RET(localNotify->Init(NotifyLoadType::DEVICE_NOTIFY));
1153 0 : CHK_RET(localNotify->SetIpc());
1154 0 : CHK_RET(localNotify->GetNotifyData(notifyInfo));
1155 0 : HCCL_INFO(
1156 : "[HcclOneSidedService][CreateAicpuNotify]resId[%llu], addr[%llu], devId[%u], tsId[%u].", notifyInfo.resId,
1157 : notifyInfo.addr, notifyInfo.devId, notifyInfo.tsId);
1158 0 : return HCCL_SUCCESS;
1159 : }
1160 :
1161 0 : u64 HcclOneSidedService::CalcTilingDynamicDataSize(HcclCMDType cmdType, u32 descNum)
1162 : {
1163 0 : u64 dynamicDataSize = 0ULL;
1164 0 : if (cmdType == HcclCMDType::HCCL_CMD_BATCH_GET || cmdType == HcclCMDType::HCCL_CMD_BATCH_PUT) {
1165 0 : dynamicDataSize = sizeof(OpTilingOneSideCommDataDes) + sizeof(HcclOneSideOpDescParam) * (descNum + 1); // signal
1166 : }
1167 0 : return dynamicDataSize;
1168 : }
1169 :
1170 0 : HcclResult HcclOneSidedService::InitAicpuTilingDataBuf(
1171 : const AicpuOneSideCommTiling& tilingInfo, u32 remoteRankId, const std::shared_ptr<HcclOneSidedConn>& conn,
1172 : const HcclOneSideOpDesc* desc, u32 descNum, u64 dynamicDataSize)
1173 : {
1174 0 : const u64 tilingDataSize = sizeof(struct OpTilingData) + dynamicDataSize;
1175 0 : if (tilingDataMem_.ptr() == nullptr) {
1176 0 : tilingDataMem_ = HostMem::alloc(std::max(tilingDataSize, TILINGDATA_BUF_SIZE));
1177 0 : CHK_PRT_RET(
1178 : tilingDataMem_.ptr() == nullptr, HCCL_ERROR("[InitAicpuTilingDataBuf] Alloc tilingDataMem failed!"),
1179 : HCCL_E_MEMORY);
1180 : }
1181 :
1182 0 : if (tilingDataSize > tilingDataMem_.size()) {
1183 0 : HCCL_INFO(
1184 : "[InitAicpuTilingDataBuf] Increase tilingDataMem from size[%llu] to tilingDataSize[%llu]",
1185 : tilingDataMem_.size(), tilingDataSize);
1186 0 : tilingDataMem_.free();
1187 0 : tilingDataMem_ = HostMem::alloc(tilingDataSize);
1188 0 : CHK_PRT_RET(
1189 : tilingDataMem_.ptr() == nullptr,
1190 : HCCL_ERROR(
1191 : "[InitAicpuTilingDataBuf] Increase tilingDataMem to "
1192 : "tilingDataSize[%llu] failed!",
1193 : tilingDataSize),
1194 : HCCL_E_MEMORY);
1195 : }
1196 :
1197 0 : const HcclCMDType cmdType = tilingInfo.cmdType;
1198 0 : HCCL_DEBUG(
1199 : "[InitAicpuTilingDataBuf] [%s] tilingDataSize[%llu] dynamicDataSize[%llu] desc[%p] descNum[%u] "
1200 : "cmdType[%u] tilingDataMem[%p] tilingDataMem.size[%llu]",
1201 : tilingInfo.tag.c_str(), tilingDataSize, dynamicDataSize, desc, descNum, cmdType, tilingDataMem_.ptr(),
1202 : tilingDataMem_.size());
1203 :
1204 : // 填充固定内容
1205 0 : HostMem tilingDataMem = tilingDataMem_.range(0, tilingDataSize);
1206 0 : CHK_PTR_NULL(tilingDataMem.ptr());
1207 0 : struct OpTilingData* tilingData = static_cast<struct OpTilingData*>(tilingDataMem.ptr());
1208 0 : CHK_SAFETY_FUNC_RET(
1209 : memcpy_s(tilingData->tag, sizeof(tilingData->tag), tilingInfo.tag.c_str(), tilingInfo.tag.length() + 1));
1210 0 : tilingData->floatOverflowMode = tilingInfo.floatOverflowMode;
1211 0 : tilingData->dumpDebug = tilingInfo.dumpDebug;
1212 0 : tilingData->debugMode = 0;
1213 0 : tilingData->srcRank = localRankInfo_.userRank;
1214 0 : tilingData->dstRank = remoteRankId;
1215 0 : tilingData->opType = static_cast<u8>(tilingInfo.cmdType);
1216 0 : tilingData->length = dynamicDataSize;
1217 0 : tilingData->customDataLength = 0;
1218 :
1219 : // 填充动态内容
1220 0 : HostMem dynamicDataMem = tilingDataMem_.range(sizeof(struct OpTilingData), dynamicDataSize);
1221 0 : CHK_PTR_NULL(dynamicDataMem.ptr());
1222 0 : auto* vDataPtr = reinterpret_cast<struct OpTilingOneSideCommDataDes*>(dynamicDataMem.ptr());
1223 0 : vDataPtr->commResParaAddr = reinterpret_cast<u64>(commResParaDevice_.ptr());
1224 0 : vDataPtr->commResParaSize = commResParaDevice_.size();
1225 0 : vDataPtr->rankSize = rankTable_->rankNum;
1226 0 : vDataPtr->linkTimeout = 0; // deprecated; 改成在AICPU侧使用qpInfo里的配置计算
1227 0 : vDataPtr->descNum = descNum + 1; // signal
1228 0 : vDataPtr->descDataLen = sizeof(HcclOneSideOpDescParam) * vDataPtr->descNum;
1229 : vDataPtr->linkType
1230 0 : = tilingInfo.useRdma ? static_cast<u8>(LinkType::LINK_ROCE) : static_cast<u8>(LinkType::LINK_HCCS);
1231 0 : if (conn != nullptr && desc != nullptr) {
1232 0 : vDataPtr->finalize = false;
1233 : auto* descParam = reinterpret_cast<HcclOneSideOpDescParam*>(
1234 0 : reinterpret_cast<u8*>(dynamicDataMem.ptr()) + sizeof(OpTilingOneSideCommDataDes));
1235 0 : CHK_RET(conn->GetTransInfo(
1236 : descParam, desc, vDataPtr->descNum, vDataPtr->transportDataAddr, vDataPtr->transportDataSize));
1237 0 : CHK_RET(hrtMemSyncCopy(
1238 : commResParaDevice_.ptr(), commResParaDevice_.size(), reinterpret_cast<void*>(&commResPara_),
1239 : sizeof(commResPara_), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
1240 : } else {
1241 0 : vDataPtr->finalize = true;
1242 : }
1243 :
1244 0 : return HCCL_SUCCESS;
1245 0 : }
1246 :
1247 0 : HcclResult HcclOneSidedService::AicpuKernelLaunch(
1248 : const std::shared_ptr<HcclOneSidedConn>& conn, const std::string& kernelName,
1249 : const AicpuOneSideCommTiling& tilingInfo, u64 tilingDataSize)
1250 : {
1251 0 : const u64 beginTime = hrtMsprofSysCycleTime();
1252 0 : std::string profName = GetCMDTypeEnumStr(tilingInfo.cmdType);
1253 0 : if (profName == "Invalid HcclCMDType" || profName == "invalid") {
1254 0 : profName = "HcclOneSideOpAicpuKernel";
1255 : } else {
1256 0 : profName += "AicpuKernel";
1257 : }
1258 :
1259 0 : s32 streamId = 0;
1260 0 : Stream mainStream = Stream(tilingInfo.stream);
1261 0 : CHK_RET(hrtGetStreamId(mainStream.ptr(), streamId));
1262 0 : HCCL_DEBUG("[%s] profName[%s] streamId[%d]", __func__, profName.c_str(), streamId);
1263 :
1264 0 : Stream launchStream = Stream(tilingInfo.stream); // 在用户流展开
1265 0 : if (!isContextLaunched_) {
1266 0 : CHK_RET(ReportProfilingCommInfo(launchStream, execStream_));
1267 0 : isContextLaunched_ = true;
1268 : }
1269 :
1270 0 : HostMem tilingDataMem = tilingDataMem_.range(0, tilingDataSize);
1271 0 : CHK_RET(AicpuUnfoldKernelLaunchV2(kernelName, tilingDataMem.ptr(), tilingDataSize, launchStream.ptr()));
1272 :
1273 : // 省略下发流，在用户流展开，已经可以和用户流任务保序，不再需要前置Post/Wait，否则会导致Kernel任务不能边展开边执行
1274 0 : HCCL_DEBUG(
1275 : "[AicpuKernelLaunch] launch in user[%u] stream[%u], launchStream[%u], execStream[%u]",
1276 : (mainStream.id() == launchStream.id()), mainStream.id(), launchStream.id(), execStream_.id());
1277 :
1278 0 : const u64 endTime = hrtMsprofSysCycleTime();
1279 0 : const s32 threadId = SalGetTid();
1280 0 : CHK_RET(ProfilingManagerPub::CallMsprofReportNodeInfo(beginTime, endTime, profName, threadId));
1281 :
1282 0 : if (conn != nullptr) {
1283 0 : if (tilingInfo.useRdma) {
1284 0 : CHK_RET(conn->WaitOpFence(tilingInfo.stream));
1285 : } else {
1286 0 : CHK_RET(LocalNotify::Wait(mainStream, dispatcher_, localAicpuNotify_[1], INVALID_VALUE_STAGE));
1287 : }
1288 : }
1289 :
1290 0 : HCCL_INFO(
1291 : "[HcclOneSidedService][AicpuKernelLaunch] exec succ, conn[%p], streamId[%u]. time[%u]", conn.get(),
1292 : mainStream.id(), (endTime - beginTime));
1293 :
1294 0 : return HCCL_SUCCESS;
1295 0 : }
1296 :
1297 0 : HcclResult HcclOneSidedService::AicpuUnfoldKernelLaunchV2(
1298 : const std::string& kernelName, void* tilingDataPtr, u64 tilingDataSize, const rtStream_t stream)
1299 : {
1300 0 : u64 commContext = 0ULL;
1301 0 : u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
1302 : std::numeric_limits<uint16_t>::max() :
1303 : NOTIFY_DEFAULT_WAIT_TIME;
1304 0 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET
1305 0 : || CommConfiger::GetInstance().GetCommConfigExecTimeOutSet(identifier_)) {
1306 0 : s32 execTimeOut = CommConfiger::GetInstance().GetCommConfigExecTimeOut(identifier_);
1307 0 : if (execTimeOut >= MAX_VALUE_U16) {
1308 0 : timeOut = MAX_VALUE_U16;
1309 : } else {
1310 0 : timeOut = execTimeOut;
1311 : }
1312 : }
1313 :
1314 0 : if (tilingDataSize > std::numeric_limits<uint32_t>::max()) {
1315 0 : HCCL_ERROR(
1316 : "[AicpuUnfoldKernelLaunchV2] tilingDataSize[%llu] exceeds the "
1317 : "maximum allowed value for u32 [%u].",
1318 : tilingDataSize, std::numeric_limits<uint32_t>::max());
1319 0 : return HCCL_E_RUNTIME;
1320 : }
1321 :
1322 0 : CHK_RET(AicpuAclKernelLaunchV2(
1323 : stream, reinterpret_cast<void*>(&commContext), sizeof(commContext), binHandle_, kernelName, false, timeOut,
1324 : tilingDataPtr, tilingDataSize, identifier_));
1325 0 : HCCL_DEBUG("[HcclOneSidedService][AicpuUnfoldKernelLaunchV2] exec succ.");
1326 0 : return HCCL_SUCCESS;
1327 : }
1328 : } // namespace hccl
|
