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