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 <climits>
12 : #include "hccl_one_sided_service.h"
13 : #include "communicator_impl.h"
14 : #include "virtual_topo.h"
15 : #include "alg_topo_package_helper.h"
16 : #include "aicpu_res_package_helper.h"
17 : #include "hccl_mem.h"
18 : #include "exception_util.h"
19 : #include "env_config.h"
20 :
21 : namespace Hccl {
22 : using namespace std;
23 :
24 : // 设置最大注册内存数量为256
25 : constexpr u32 maxregisteredMem = 256;
26 :
27 16 : static void OneSidedSetModuleDataName(ModuleData &module, const std::string &name)
28 : {
29 16 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
30 16 : if (ret != 0) {
31 0 : THROW<InternalException>(StringFormat("strcpy_s name %s failed. ret[%d]", name.c_str(), ret));
32 : }
33 16 : }
34 :
35 : template <class T, class U> u16 CalcFieldOffset(T *target, U *base)
36 : {
37 : return static_cast<u16>(static_cast<const char *>(static_cast<void *>(target))
38 : - static_cast<const char *>(static_cast<void *>(base)));
39 : }
40 :
41 :
42 58 : HcclOneSidedService::HcclOneSidedService(CommunicatorImpl &comm) : comm_(&comm)
43 : {
44 58 : AddOpCounterMems();
45 58 : }
46 :
47 58 : HcclOneSidedService::~HcclOneSidedService()
48 : {
49 60 : for (const auto &pair : desc2netDevMap_) {
50 2 : const HcclNetDev &hcclNetDev = pair.second;
51 2 : HcclResult ret = HcclNetDevClose(hcclNetDev);
52 2 : if (ret != HCCL_SUCCESS) {
53 0 : HCCL_ERROR("[HcclOneSidedService][~HcclOneSidedService]HcclNetDevClose failed, descStr[%s], ret[%d].",
54 : pair.first.c_str(), ret);
55 : }
56 : }
57 58 : }
58 :
59 56 : LinkData HcclOneSidedService::GetLinkData(RankId remoteRankId)
60 : {
61 56 : if (linkDataMap_.find(remoteRankId) == linkDataMap_.end()) {
62 : // 组建linkData
63 0 : LinkData linkData(comm_->GetRankGraph()->GetPaths(0, comm_->GetMyRank(), remoteRankId)[0]);
64 0 : linkDataMap_.emplace(remoteRankId, linkData);
65 : }
66 168 : HCCL_INFO("[HcclOneSidedService][GetLinkData] linkData[%s]", linkDataMap_.at(remoteRankId).Describe().c_str());
67 56 : return linkDataMap_.at(remoteRankId);
68 : }
69 :
70 53 : HcclResult HcclOneSidedService::CheckLink(LinkData linkData) const
71 : {
72 159 : HCCL_INFO("[HcclOneSidedService][CheckLink] linkData[%s]", linkData.Describe().c_str());
73 206 : CHK_PRT_RET(
74 : (linkData.GetLinkProtocol() != LinkProtocol::UB_CTP && linkData.GetLinkProtocol() != LinkProtocol::UB_TP),
75 : HCCL_ERROR("[HcclOneSidedService][CheckLink] Proto is not UB, not support"), HCCL_E_NOT_SUPPORT);
76 2 : CHK_PRT_RET(linkData.GetHop() > 1, HCCL_ERROR("[HcclOneSidedService][CheckLink]Hop is greater than 1, not support"),
77 : HCCL_E_NOT_SUPPORT);
78 2 : return HCCL_SUCCESS;
79 : }
80 :
81 4 : HcclResult HcclOneSidedService::RegMem(void *addr, u64 size, HcclMemType type, RankId remoteRankId,
82 : HcclMemDesc &localMemDesc)
83 : {
84 4 : CHK_PTR_NULL(addr);
85 10 : CHK_PRT_RET(type == HcclMemType::HCCL_MEM_TYPE_HOST,
86 : HCCL_ERROR("[HcclOneSidedService][RegMem]HCCL_MEM_TYPE_HOST is not supported"), HCCL_E_NOT_SUPPORT);
87 6 : HCCL_INFO("[HcclOneSidedService][RegMem]addr[%p], size[%llu], type[%d], remoteRankId[%u]", addr, size, type,
88 : remoteRankId);
89 2 : LinkData linkData = GetLinkData(remoteRankId);
90 2 : RmaMemDesc *localRmaMemDesc = static_cast<RmaMemDesc *>(static_cast<void *>(localMemDesc.desc));
91 2 : CHK_PTR_NULL(localRmaMemDesc);
92 2 : CHK_PRT_RET(registeredMemCnt_ >= maxregisteredMem,
93 : HCCL_ERROR("[HcclOneSidedService][RegMem]registered memory counts=[%u] exceeds limit[%u]", registeredMemCnt_,
94 : maxregisteredMem),
95 : HCCL_E_UNAVAIL);
96 2 : HcclNetDevInfos info;
97 2 : info.addr.protoType = HcclNetDevice::ConvertHcclProtoToLinkProto(linkData.GetLocalPort().GetProto());
98 2 : info.addr.type = HCCL_ADDR_TYPE_IP_V4;
99 2 : info.netdevDeployment = HcclNetDevice::ConvertDeploymentType(linkData.GetLocalPort().GetType());
100 2 : info.devicePhyId = comm_->GetDevicePhyId();
101 2 : info.addr.addr = linkData.GetLocalPort().GetAddr().GetBinaryAddress().addr;
102 : HcclNetDev netDev;
103 2 : HcclResult ret = HcclNetDevOpen(&info, &netDev);
104 2 : if (ret != HCCL_SUCCESS) {
105 0 : HCCL_ERROR("[HcclOneSidedService][RegMem]HcclNetDevOpen failed, ret[%d].", ret);
106 0 : return ret;
107 : }
108 2 : HcclMem localMem{type, addr, size};
109 : HcclBuf buf;
110 2 : ret = HcclMemReg(netDev, &localMem, &buf);
111 2 : if ((ret != HCCL_SUCCESS) && (ret != HCCL_E_AGAIN)) {
112 0 : HCCL_ERROR("[HcclOneSidedService][RegMem]HcclMemReg failed, ret[%d].", ret);
113 0 : CHK_RET(HcclNetDevClose(netDev));
114 0 : return ret;
115 : }
116 : string logInfo = ret == HCCL_SUCCESS ? "Register memory success!"
117 2 : : "Memory is already registered, just increase the reference count.";
118 6 : HCCL_INFO("[HcclOneSidedService][RegMem]:%s Add key {%p, %llu}", logInfo.c_str(), addr, size);
119 2 : localRmaMemDesc->localRankId = comm_->GetMyRank();
120 2 : localRmaMemDesc->remoteRankId = remoteRankId;
121 2 : char *desc = localRmaMemDesc->memDesc;
122 2 : uint64_t descLen = 0;
123 2 : ret = HcclMemExport(&buf, &desc, &descLen);
124 2 : if (ret != HCCL_SUCCESS) {
125 0 : HCCL_ERROR("[HcclOneSidedService][RegMem]HcclMemExport failed, ret[%d]", ret);
126 0 : CHK_RET(HcclNetDevClose(netDev));
127 0 : return ret;
128 : }
129 2 : registeredMemCnt_++;
130 2 : std::string descStr(localRmaMemDesc->memDesc, TRANSPORT_EMD_ESC_SIZE);
131 2 : desc2HcclBufMapLocalUb_.emplace(descStr, buf);
132 2 : desc2netDevMap_.emplace(descStr, netDev);
133 2 : return HCCL_SUCCESS;
134 2 : }
135 :
136 4 : HcclResult HcclOneSidedService::DeregMem(const HcclMemDesc &localMemDesc)
137 : {
138 : // 若当前内存注册数量为0，则返回找不到内存
139 4 : if (registeredMemCnt_ == 0) {
140 6 : HCCL_ERROR("[HcclOneSidedService][DeregMem]Registered memory is 0, please register first.");
141 2 : return HCCL_E_NOT_FOUND;
142 : }
143 2 : const RmaMemDesc *localRmaMemDesc = static_cast<const RmaMemDesc *>(static_cast<const void *>(localMemDesc.desc));
144 2 : std::string descStr(localRmaMemDesc->memDesc, TRANSPORT_EMD_ESC_SIZE);
145 2 : if (desc2HcclBufMapLocalUb_.find(descStr) == desc2HcclBufMapLocalUb_.end()) {
146 0 : HCCL_ERROR("[HcclOneSidedService][GetHcclBufByDesc]memory is not registered, please register first.");
147 0 : return HCCL_E_NOT_FOUND;
148 : }
149 2 : HcclBuf buf = desc2HcclBufMapLocalUb_.at(descStr);
150 2 : HcclResult ret = HcclMemDereg(&buf);
151 2 : if (ret == HCCL_SUCCESS) {
152 2 : registeredMemCnt_--;
153 2 : desc2HcclBufMapLocalUb_.erase(descStr);
154 : }
155 2 : if (desc2netDevMap_.find(descStr) == desc2netDevMap_.end()) {
156 0 : HCCL_ERROR("[HcclOneSidedService][GetHcclBufByDesc]NetDev is not open, please register first.");
157 0 : return HCCL_E_INTERNAL;
158 : }
159 2 : return HCCL_SUCCESS;
160 2 : }
161 :
162 1 : HcclResult HcclOneSidedService::CreateConnection(std::shared_ptr<HcclOneSidedConn> &tempConn, LinkData linkData)
163 : {
164 1 : if (isOpModeReady_ == false) {
165 1 : CHK_RET(comm_->RecoverOpMode(1));
166 1 : isOpModeReady_ = true;
167 : }
168 3 : HCCL_INFO("[HcclOneSidedService][CreateConnection] start");
169 3 : HCCL_INFO("[HcclOneSidedService][CreateConnection] linkData[%s]", linkData.Describe().c_str());
170 1 : tempConn = make_shared<HcclOneSidedConn>(comm_, linkData);
171 :
172 1 : CHK_PTR_NULL(tempConn);
173 3 : HCCL_INFO("[HcclOneSidedService][CreateConnection] end");
174 1 : return HCCL_SUCCESS;
175 : }
176 :
177 56 : HcclResult HcclOneSidedService::ExchangeMemDesc(RankId remoteRankId, const HcclMemDescs &localMemDescs,
178 : HcclMemDescs &remoteMemDescs, u32 &actualNumOfRemote)
179 : {
180 56 : if (comm_->GetCommExecuteConfig().accState != AcceleratorState::AICPU_TS) {
181 6 : HCCL_ERROR("[HcclOneSidedService][%s] only support aicpu, current accelerator[%s]", __func__,
182 : comm_->GetCommExecuteConfig().accState.Describe().c_str());
183 2 : return HCCL_E_NOT_SUPPORT;
184 : }
185 54 : comm_->SetOpExecuteConfig(comm_->GetCommExecuteConfig());
186 : // 组装linkData
187 54 : LinkData linkData = GetLinkData(remoteRankId);
188 :
189 162 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] Find HcclOneSidedConn");
190 54 : shared_ptr<HcclOneSidedConn> tempConn;
191 : // 查找是否已存在对端连接，不存在则创建
192 54 : std::unique_lock oneSidedConnslock(oneSidedConnsMutex_);
193 54 : auto it = oneSidedConns_.find(remoteRankId);
194 54 : if (it == oneSidedConns_.end()) {
195 : // 检测对端是否符合建链要求
196 206 : CHK_RET(CheckLink(linkData));
197 : // 创建Conn对象
198 5 : CHK_RET(CreateConnection(tempConn, linkData));
199 1 : oneSidedConns_.emplace(remoteRankId, tempConn);
200 : } else {
201 1 : tempConn = it->second;
202 : }
203 :
204 2 : CHK_PTR_NULL(tempConn);
205 6 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] tempConn linkData[%s]", linkData.Describe().c_str());
206 6 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] ExchangeMemDesc");
207 2 : return tempConn->ExchangeMemDesc(localMemDescs, remoteMemDescs, actualNumOfRemote);
208 54 : }
209 :
210 2 : HcclResult HcclOneSidedService::EnableMemAccess(const HcclMemDesc &remoteMemDesc, HcclMem &remoteMem)
211 : {
212 2 : CHK_PTR_NULL(remoteMemDesc.desc);
213 : // 将HcclMemDesc转化为RmaMemDesc
214 2 : const RmaMemDesc *remoteRmaMemDesc = static_cast<const RmaMemDesc *>(static_cast<const void *>(remoteMemDesc.desc));
215 2 : RankId remoteRankId = remoteRmaMemDesc->localRankId;
216 :
217 6 : HCCL_INFO("[HcclOneSidedService][EnableMemAccess] Get remoteRankId[%u]", remoteRankId);
218 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
219 2 : if (oneSidedConns_.find(remoteRankId) == oneSidedConns_.end()) {
220 3 : HCCL_ERROR("[HcclOneSidedService][EnableMemAccess]connection not found, remoteRank[%u].", remoteRankId);
221 1 : return HCCL_E_NOT_FOUND;
222 : }
223 :
224 3 : HCCL_INFO("[HcclOneSidedService][EnableMemAccess] EnableMemAccess.");
225 1 : oneSidedConns_.at(remoteRankId)->EnableMemAccess(remoteMemDesc, remoteMem);
226 1 : return HCCL_SUCCESS;
227 2 : }
228 :
229 2 : HcclResult HcclOneSidedService::DisableMemAccess(const HcclMemDesc &remoteMemDesc)
230 : {
231 2 : CHK_PTR_NULL(remoteMemDesc.desc);
232 : // 将HcclMemDesc转化为RmaMemDesc
233 2 : const RmaMemDesc *remoteRmaMemDesc = static_cast<const RmaMemDesc *>(static_cast<const void *>(remoteMemDesc.desc));
234 :
235 : // 获取Conn对象
236 2 : RankId remoteRankId = remoteRmaMemDesc->localRankId;
237 6 : HCCL_INFO("[HcclOneSidedService][DisableMemAccess] Get remoteRankId[%u]", remoteRankId);
238 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
239 2 : if (oneSidedConns_.find(remoteRankId) == oneSidedConns_.end()) {
240 3 : HCCL_ERROR("[HcclOneSidedService][DisableMemAccess]connection not found by remoteRankId[%u].", remoteRankId);
241 1 : return HCCL_E_NOT_FOUND;
242 : }
243 :
244 3 : HCCL_INFO("[HcclOneSidedService][DisableMemAccess] DisableMemAccess");
245 1 : oneSidedConns_.at(remoteRankId)->DisableMemAccess(remoteMemDesc);
246 1 : return HCCL_SUCCESS;
247 2 : }
248 :
249 2 : HcclResult HcclOneSidedService::BatchPutGetDevBufs(const HcclOneSideOpDesc *desc, u32 descNum,
250 : std::shared_ptr<HcclOneSidedConn> oneSidedConn)
251 : {
252 4 : vector<HcclAicpuLocBufLite> hostBatchPutGetLocalBufferSliceBufs(descNum);
253 2 : vector<HcclAicpuLocBufLite> hostBatchPutGetRemoteBufferSliceBufs(descNum);
254 2 : CHK_RET(oneSidedConn->BatchBufferSlice(desc, descNum, hostBatchPutGetLocalBufferSliceBufs,
255 : hostBatchPutGetRemoteBufferSliceBufs));
256 :
257 2 : devBatchPutGetLocalBufs = make_shared<DevBuffer>(sizeof(HcclAicpuLocBufLite) * descNum);
258 2 : devBatchPutGetRemoteBufs = make_shared<DevBuffer>(sizeof(HcclAicpuLocBufLite) * descNum);
259 :
260 2 : HrtMemcpy(reinterpret_cast<void *>(devBatchPutGetLocalBufs->GetAddr()), devBatchPutGetLocalBufs->GetSize(),
261 2 : static_cast<void *>(hostBatchPutGetLocalBufferSliceBufs.data()), sizeof(HcclAicpuLocBufLite) * descNum,
262 : RT_MEMCPY_HOST_TO_DEVICE);
263 :
264 2 : HrtMemcpy(reinterpret_cast<void *>(devBatchPutGetRemoteBufs->GetAddr()), devBatchPutGetRemoteBufs->GetSize(),
265 2 : static_cast<void *>(hostBatchPutGetRemoteBufferSliceBufs.data()), sizeof(HcclAicpuLocBufLite) * descNum,
266 : RT_MEMCPY_HOST_TO_DEVICE);
267 :
268 2 : return HCCL_SUCCESS;
269 2 : }
270 :
271 2 : std::vector<char> HcclOneSidedService::PackOpData(const CollAlgOpReq &req) const
272 : {
273 2 : std::vector<ModuleData> dataVec;
274 2 : dataVec.resize(AicpuResMgrType::__COUNT__);
275 :
276 2 : AicpuResMgrType resType = AicpuResMgrType::STREAM;
277 2 : OneSidedSetModuleDataName(dataVec[resType], "StreamManager");
278 2 : dataVec[resType].data = comm_->GetAicpuStreamManager().GetPackedData();
279 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
280 :
281 2 : resType = AicpuResMgrType::QUEUE_NOTIFY;
282 2 : OneSidedSetModuleDataName(dataVec[resType], "QueueNotifyManager");
283 2 : dataVec[resType].data = comm_->GetAicpuQueueNotifyManager().GetPackedData();
284 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
285 :
286 2 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
287 2 : OneSidedSetModuleDataName(dataVec[resType], "QueueWaitGroupCntNotifyManager");
288 2 : dataVec[resType].data = comm_->GetQueueWaitGroupCntNotifyManager().GetPackedData();
289 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
290 :
291 2 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
292 2 : OneSidedSetModuleDataName(dataVec[resType], "GetBcastPostCntNotifyManager");
293 2 : dataVec[resType].data = comm_->GetBcastPostCntNotifyManager().GetPackedData();
294 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
295 :
296 2 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
297 2 : OneSidedSetModuleDataName(dataVec[resType], "HostDeviceSyncNotifyManager");
298 2 : dataVec[resType].data = comm_->GetHostDeviceSyncNotifyManager().GetPackedData();
299 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
300 :
301 2 : resType = AicpuResMgrType::TRANSPORT;
302 2 : OneSidedSetModuleDataName(dataVec[resType], "MemTransportManager");
303 2 : auto op = comm_->GetCurrentCollOperator();
304 : // GetOpbasedPackedData由于单边通信隔离，会找不到Transport
305 2 : if (op->opMode == OpMode::OPBASE) { // 单算子模式
306 2 : dataVec[resType].data = comm_->GetMemTransportManager()->GetOneSidedPackedData();
307 : } else {
308 0 : THROW<InternalException>(StringFormat("opMode=%s failed", op->opMode.Describe().c_str()));
309 : }
310 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
311 :
312 2 : resType = AicpuResMgrType::ALG_TOPO;
313 2 : OneSidedSetModuleDataName(dataVec[resType], req.algName);
314 : AlgTopoPackageHelper algTopoHelper;
315 2 : dataVec[resType].data = algTopoHelper.GetPackedData(req.resReq.topoInfo);
316 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
317 :
318 2 : resType = AicpuResMgrType::CONNECTD_MGR;
319 2 : OneSidedSetModuleDataName(dataVec[resType], "ConnectedManager");
320 2 : dataVec[resType].data = comm_->GetRankGraph()->GetPackedData(req.resReq.levelRankPairs);
321 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
322 :
323 : AicpuResPackageHelper helper;
324 4 : return helper.GetPackedData(dataVec);
325 2 : }
326 :
327 2 : void HcclOneSidedService::FillOneSidedOperator(OpType type, RankId remoteRankId, const HcclOneSideOpDesc *desc) const
328 : {
329 2 : CollOpParams opParams;
330 :
331 2 : opParams.dataType = HcclDataTypeToDataType(desc->dataType);
332 2 : opParams.count = desc->count;
333 :
334 : // sendBuf/recvBuf当前不使用，等待后续扩展
335 2 : opParams.sendBuf = desc->localAddr;
336 2 : opParams.recvBuf = desc->remoteAddr;
337 :
338 2 : opParams.opType = type;
339 2 : opParams.dstRank = remoteRankId;
340 2 : std::string opTag = comm_->GetId();
341 :
342 6 : HCCL_INFO(
343 : "[HcclOneSidedService][FillOneSidedOperator] CovertToCurrentCollOperator opType[%s], dstRank[%u], opTag[%s]",
344 : opParams.opType.Describe().c_str(), opParams.dstRank, opTag.c_str());
345 2 : comm_->CovertToCurrentCollOperator(opTag, opParams, OpMode::OPBASE);
346 2 : }
347 :
348 2 : void HcclOneSidedService::AddPostToUserStream(const Stream &stream) const
349 : {
350 2 : auto postNotify = comm_->GetHostDeviceSyncNotifyManager().GetDeviceWaitNotify();
351 :
352 2 : postNotify->Post(stream);
353 2 : }
354 :
355 2 : void HcclOneSidedService::AddWaitToUserStream(const Stream &stream) const
356 : {
357 2 : auto waitNotify = comm_->GetHostDeviceSyncNotifyManager().GetHostWaitNotify();
358 :
359 2 : waitNotify->Wait(stream, 1000); // host 和 device sync流程，等待1000ms
360 2 : }
361 :
362 2 : void HcclOneSidedService::SetOneSidedKernelLaunchParam(HcclKernelLaunchParam ¶m, const DevBuffer *mem) const
363 : {
364 2 : CollOperator op = *comm_->GetCurrentCollOperator();
365 :
366 6 : HCCL_INFO("[HcclOneSidedService][SetOneSidedKernelLaunchParam] op.opType[%s]", op.opType.Describe().c_str());
367 2 : param.kernel.comm.idIndex = comm_->GetIdIndex();
368 2 : param.kernel.comm.myRank = comm_->GetMyRank();
369 2 : param.kernel.comm.rankSize = comm_->GetRankSize();
370 2 : param.kernel.comm.devType = comm_->GetDevType();
371 2 : param.kernel.comm.devPhyId = comm_->GetDevicePhyId();
372 2 : param.kernel.comm.opCounterAddr = static_cast<u64>(counterBuf->GetAddr());
373 2 : auto ret = strcpy_s(param.kernel.comm.commId, sizeof(param.kernel.comm.commId), comm_->GetId().data());
374 2 : if (ret != EOK) {
375 0 : THROW<InternalException>(
376 0 : StringFormat("HcclOneSidedService::SetOneSidedKernelLaunchParam, strcpy_s commId failed! ret[%d]", ret));
377 : }
378 :
379 2 : param.kernel.oneSidedComm = true;
380 :
381 2 : param.kernel.op.algOperator.opMode = op.opMode;
382 2 : param.kernel.op.algOperator.opType = op.opType;
383 :
384 2 : param.kernel.binaryResAddr = mem->GetAddr();
385 2 : param.kernel.binaryResSize = mem->GetSize();
386 :
387 2 : param.kernel.op.sendRecvRemoteRank = op.sendRecvRemoteRank;
388 :
389 2 : param.kernel.kfcControlTransferH2DParams = comm_->GetKfcControlTransferH2D().GetCommunicateParams();
390 2 : param.kernel.kfcControlTransferD2HParams = comm_->GetKfcStatusTransferD2H().GetCommunicateParams();
391 2 : }
392 :
393 2 : void HcclOneSidedService::OneSidedAicpuKernelLaunch(HcclKernelLaunchParam ¶m, Stream &stream) const
394 : {
395 2 : const aclrtFuncHandle funcHandle = comm_->GetAicpuKernelFuncHandle(param.kernelName);
396 2 : constexpr u32 numBlocks = 1;
397 : aclrtLaunchKernelCfg cfg;
398 : aclrtLaunchKernelAttr attr;
399 2 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
400 2 : auto timeoutCheck = EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
401 : // aicpu kernal超时时间: X+30s
402 2 : attr.value.timeout = static_cast<u16>((timeoutCheck == 0) ? timeoutCheck : (timeoutCheck + 30));
403 2 : cfg.numAttrs = 1;
404 2 : cfg.attrs = &attr;
405 2 : AddPostToUserStream(stream);
406 6 : HCCL_INFO("[HcclOneSidedService::AicpuKernelLaunch] param.soName: %s, param.kernelName: %s", param.soName,
407 : param.kernelName);
408 2 : HrtAicpuLaunchKernelWithHostArgs(
409 : funcHandle, numBlocks,
410 2 : comm_->GetAicpuStreamManager().GetFreeStream()->GetPtr(), &cfg,
411 2 : ¶m.kernel, sizeof(HcclKernelParamLite));
412 6 : HCCL_INFO("[HcclOneSidedService][AicpuKernelLaunch] param.kernel.algName: %s HrtAicpuLaunchKernelWithHostArgs end!",
413 : param.kernel.algName);
414 2 : AddWaitToUserStream(stream);
415 2 : }
416 :
417 2 : DevBuffer *HcclOneSidedService::PackResToKernelLanuch(CollAlgOpReq &opReq)
418 : {
419 2 : auto it = OneSidedLoadMap.find(opReq.algName);
420 2 : if (it != OneSidedLoadMap.end()) { // 已经向Device Mem写过资源
421 0 : HCCL_INFO("[OpBasedCollProcess] tag[%s] devMem has been allocated, reuse it", opReq.algName.c_str());
422 0 : return it->second.get();
423 : }
424 :
425 6 : HCCL_INFO("[HcclOneSidedService][PackResToKernelLanuch], PackOpData start");
426 : // 打包单边通信资源信息到device
427 2 : auto buffer = PackOpData(opReq);
428 2 : shared_ptr<DevBuffer> devMem = make_shared<DevBuffer>(buffer.size()); // 申请device内存
429 :
430 6 : HCCL_INFO("[HcclOneSidedService][PackResToKernelLanuch], HrtMemSyncCopy start");
431 2 : HrtMemcpy(reinterpret_cast<void *>(devMem->GetAddr()), devMem->GetSize(), buffer.data(), buffer.size(),
432 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到device内存
433 6 : HCCL_INFO("HcclOneSidedService::BatchGet PackOpData: PackedData %s",
434 : Bytes2hex(buffer.data(), buffer.size()).c_str());
435 2 : OneSidedLoadMap.insert(make_pair(opReq.algName, devMem));
436 :
437 2 : return devMem.get();
438 2 : }
439 :
440 2 : HcclResult HcclOneSidedService::BatchOpKernelLaunch(OpType opType, RankId remoteRankId, const HcclOneSideOpDesc *desc,
441 : u32 descNum, shared_ptr<Stream> stream)
442 : {
443 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] start");
444 2 : comm_->GetAicpuStreamManager().AllocFreeStream();
445 6 : HCCL_INFO("[HcclOneSidedService][AllocStreams] start");
446 2 : comm_->GetAicpuStreamManager().AllocStreams(1);
447 :
448 2 : CHK_PTR_NULL(desc);
449 6 : HCCL_INFO(
450 : "[HcclOneSidedService][BatchOpKernelLaunch] desc: localAddr:[%p],remoteAddr:[%p],count:[%llu],dataType:[%d]",
451 : desc->localAddr, desc->remoteAddr, desc->count, desc->dataType);
452 :
453 2 : CollAlgOpReq opReq;
454 2 : opReq.algName = OpTypeToString(opType);
455 2 : opReq.resReq.levelRankPairs.push_back(make_pair(0, remoteRankId));
456 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] FillOneSidedOperator start");
457 : // 填充通信算子信息
458 2 : FillOneSidedOperator(opType, remoteRankId, desc);
459 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] PackResToKernelLanuch start");
460 : // 打包device展开资源信息
461 2 : DevBuffer *devMem = PackResToKernelLanuch(opReq);
462 : // 组kernelLaunch参数
463 2 : HcclKernelLaunchParam param;
464 :
465 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] SetOneSidedKernelLaunchParam start");
466 : // 构造单边通信公共参数
467 2 : SetOneSidedKernelLaunchParam(param, devMem);
468 :
469 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
470 2 : auto it = oneSidedConns_.find(remoteRankId);
471 2 : if (it == oneSidedConns_.end()) {
472 0 : HCCL_ERROR("[HcclMemCommunication][BatchGet] Can't find oneSidedConn by remoteRank %u", remoteRankId);
473 0 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
474 : }
475 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] BatchPutGetDevBufs start");
476 2 : CHK_RET(BatchPutGetDevBufs(desc, descNum, it->second));
477 2 : oneSidedConnslock.unlock();
478 :
479 2 : param.kernel.op.batchPutGetDescNum = descNum;
480 2 : param.kernel.op.batchPutGetLocalAddr = reinterpret_cast<void *>(devBatchPutGetLocalBufs.get()->GetAddr());
481 2 : param.kernel.op.batchPutGetRemoteAddr = reinterpret_cast<void *>(devBatchPutGetRemoteBufs.get()->GetAddr());
482 2 : auto ret = strcpy_s(param.kernel.tagKey, sizeof(param.kernel.tagKey), opReq.algName.c_str());
483 2 : if (ret != EOK) {
484 0 : THROW<InternalException>(
485 0 : StringFormat("[HcclOneSidedService][BatchOpKernelLaunch], strcpy_s opReq.algName failed! ret[%d]", ret));
486 : }
487 :
488 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] OneSidedAicpuKernelLaunch start");
489 : // 启动kernel
490 2 : OneSidedAicpuKernelLaunch(param, *stream);
491 2 : return HCCL_SUCCESS;
492 2 : }
493 :
494 774 : HcclResult HcclOneSidedService::BatchPut(RankId remoteRankId, const HcclOneSideOpDesc *desc, u32 descNum,
495 : const rtStream_t stream)
496 : {
497 2329 : HCCL_INFO("[HcclOneSidedService][BatchPut] start");
498 763 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
499 762 : auto it = oneSidedConns_.find(remoteRankId);
500 660 : if (it == oneSidedConns_.end()) {
501 2197 : HCCL_ERROR("[HcclMemCommunication][BatchPut] Can't find oneSidedConn by remoteRank %u", remoteRankId);
502 785 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
503 : }
504 1 : oneSidedConnslock.unlock();
505 :
506 3 : HCCL_INFO("[HcclOneSidedService][BatchPut] BatchOpKernelLaunch start");
507 1 : CHK_RET(BatchOpKernelLaunch(OpType::BATCHPUT, remoteRankId, desc, descNum, std::make_shared<Stream>(stream)));
508 :
509 3 : HCCL_INFO("[HcclOneSidedService][BatchPut] end");
510 1 : return HCCL_SUCCESS;
511 801 : }
512 :
513 2 : HcclResult HcclOneSidedService::BatchGet(RankId remoteRankId, const HcclOneSideOpDesc *desc, u32 descNum,
514 : const rtStream_t stream)
515 : {
516 6 : HCCL_INFO("[HcclOneSidedService][BatchGet] start");
517 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
518 2 : auto it = oneSidedConns_.find(remoteRankId);
519 2 : if (it == oneSidedConns_.end()) {
520 3 : HCCL_ERROR("[HcclMemCommunication][BatchGet] Can't find oneSidedConn by remoteRank %u", remoteRankId);
521 1 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
522 : }
523 1 : oneSidedConnslock.unlock();
524 :
525 3 : HCCL_INFO("[HcclOneSidedService][BatchGet] BatchOpKernelLaunch start");
526 1 : CHK_RET(BatchOpKernelLaunch(OpType::BATCHGET, remoteRankId, desc, descNum, std::make_shared<Stream>(stream)));
527 :
528 3 : HCCL_INFO("[HcclOneSidedService][BatchGet] end");
529 1 : return HCCL_SUCCESS;
530 2 : }
531 :
532 58 : void HcclOneSidedService::AddOpCounterMems()
533 : {
534 174 : HCCL_INFO("[HcclOneSidedService::%s] start.", __func__);
535 :
536 58 : constexpr u64 FOUR_BYTES = 4;
537 58 : u64 size = FOUR_BYTES * 3; // 第一个四字节用于计数加1, 后面两个四字节分别保存headCounter和tailCounter
538 58 : counterBuf = std::make_shared<DevBuffer>(size);
539 :
540 : // 初始化第一个四字节置1, 用于计数加1, reduce task add 1
541 58 : u64 srcSize = FOUR_BYTES;
542 58 : float srcValue = 1;
543 58 : void *srcAddr = reinterpret_cast<void *>(counterBuf->GetAddr());
544 58 : HrtMemcpy(srcAddr, srcSize, &srcValue, srcSize, RT_MEMCPY_HOST_TO_DEVICE);
545 :
546 : // 初始化后面两个四字节置0
547 58 : u64 countMemSize = srcSize;
548 58 : float startValue = 0; // value为0表示从0开始计数
549 58 : void *headCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize);
550 58 : void *tailCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize * 2);
551 58 : HrtMemcpy(headCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
552 58 : HrtMemcpy(tailCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
553 :
554 174 : HCCL_INFO("[HcclOneSidedService::%s] end, counterBuf[%llu] srcAddr[%p] headCountAddr[%p] tailCountAddr[%p].", __func__,
555 : counterBuf->GetAddr(), srcAddr, headCountAddr, tailCountAddr);
556 58 : }
557 :
558 0 : DevBuffer *HcclOneSidedService::GetOpCounterBuf()
559 : {
560 0 : return counterBuf.get();
561 : }
562 : } // namespace Hccl
|
