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