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 : if (comm_->GetCommExecuteConfig().accState != AcceleratorState::AICPU_TS) {
184 6 : HCCL_ERROR(
185 : "[HcclOneSidedService][%s] only support aicpu, current accelerator[%s]", __func__,
186 : comm_->GetCommExecuteConfig().accState.Describe().c_str());
187 2 : return HCCL_E_NOT_SUPPORT;
188 : }
189 54 : comm_->SetOpExecuteConfig(comm_->GetCommExecuteConfig());
190 : // 组装linkData
191 54 : LinkData linkData = GetLinkData(remoteRankId);
192 :
193 162 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] Find HcclOneSidedConn");
194 54 : shared_ptr<HcclOneSidedConn> tempConn;
195 : // 查找是否已存在对端连接，不存在则创建
196 54 : std::unique_lock oneSidedConnslock(oneSidedConnsMutex_);
197 54 : auto it = oneSidedConns_.find(remoteRankId);
198 54 : if (it == oneSidedConns_.end()) {
199 : // 检测对端是否符合建链要求
200 206 : CHK_RET(CheckLink(linkData));
201 : // 创建Conn对象
202 5 : CHK_RET(CreateConnection(tempConn, linkData));
203 1 : oneSidedConns_.emplace(remoteRankId, tempConn);
204 : } else {
205 1 : tempConn = it->second;
206 : }
207 :
208 2 : CHK_PTR_NULL(tempConn);
209 6 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] tempConn linkData[%s]", linkData.Describe().c_str());
210 6 : HCCL_INFO("[HcclOneSidedService][ExchangeMemDesc] ExchangeMemDesc");
211 2 : return tempConn->ExchangeMemDesc(localMemDescs, remoteMemDescs, actualNumOfRemote);
212 54 : }
213 :
214 2 : HcclResult HcclOneSidedService::EnableMemAccess(const HcclMemDesc& remoteMemDesc, HcclMem& remoteMem)
215 : {
216 2 : CHK_PTR_NULL(remoteMemDesc.desc);
217 : // 将HcclMemDesc转化为RmaMemDesc
218 2 : const RmaMemDesc* remoteRmaMemDesc = static_cast<const RmaMemDesc*>(static_cast<const void*>(remoteMemDesc.desc));
219 2 : RankId remoteRankId = remoteRmaMemDesc->localRankId;
220 :
221 6 : HCCL_INFO("[HcclOneSidedService][EnableMemAccess] Get remoteRankId[%u]", remoteRankId);
222 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
223 2 : if (oneSidedConns_.find(remoteRankId) == oneSidedConns_.end()) {
224 3 : HCCL_ERROR("[HcclOneSidedService][EnableMemAccess]connection not found, remoteRank[%u].", remoteRankId);
225 1 : return HCCL_E_NOT_FOUND;
226 : }
227 :
228 3 : HCCL_INFO("[HcclOneSidedService][EnableMemAccess] EnableMemAccess.");
229 1 : oneSidedConns_.at(remoteRankId)->EnableMemAccess(remoteMemDesc, remoteMem);
230 1 : return HCCL_SUCCESS;
231 2 : }
232 :
233 2 : HcclResult HcclOneSidedService::DisableMemAccess(const HcclMemDesc& remoteMemDesc)
234 : {
235 2 : CHK_PTR_NULL(remoteMemDesc.desc);
236 : // 将HcclMemDesc转化为RmaMemDesc
237 2 : const RmaMemDesc* remoteRmaMemDesc = static_cast<const RmaMemDesc*>(static_cast<const void*>(remoteMemDesc.desc));
238 :
239 : // 获取Conn对象
240 2 : RankId remoteRankId = remoteRmaMemDesc->localRankId;
241 6 : HCCL_INFO("[HcclOneSidedService][DisableMemAccess] Get remoteRankId[%u]", remoteRankId);
242 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
243 2 : if (oneSidedConns_.find(remoteRankId) == oneSidedConns_.end()) {
244 3 : HCCL_ERROR("[HcclOneSidedService][DisableMemAccess]connection not found by remoteRankId[%u].", remoteRankId);
245 1 : return HCCL_E_NOT_FOUND;
246 : }
247 :
248 3 : HCCL_INFO("[HcclOneSidedService][DisableMemAccess] DisableMemAccess");
249 1 : oneSidedConns_.at(remoteRankId)->DisableMemAccess(remoteMemDesc);
250 1 : return HCCL_SUCCESS;
251 2 : }
252 :
253 2 : HcclResult HcclOneSidedService::BatchPutGetDevBufs(
254 : const HcclOneSideOpDesc* desc, u32 descNum, std::shared_ptr<HcclOneSidedConn> oneSidedConn)
255 : {
256 4 : vector<HcclAicpuLocBufLite> hostBatchPutGetLocalBufferSliceBufs(descNum);
257 2 : vector<HcclAicpuLocBufLite> hostBatchPutGetRemoteBufferSliceBufs(descNum);
258 2 : CHK_RET(oneSidedConn->BatchBufferSlice(
259 : desc, descNum, hostBatchPutGetLocalBufferSliceBufs, hostBatchPutGetRemoteBufferSliceBufs));
260 :
261 2 : devBatchPutGetLocalBufs = make_shared<DevBuffer>(sizeof(HcclAicpuLocBufLite) * descNum);
262 2 : devBatchPutGetRemoteBufs = make_shared<DevBuffer>(sizeof(HcclAicpuLocBufLite) * descNum);
263 :
264 2 : HrtMemcpy(
265 4 : reinterpret_cast<void*>(devBatchPutGetLocalBufs->GetAddr()), devBatchPutGetLocalBufs->GetSize(),
266 2 : static_cast<void*>(hostBatchPutGetLocalBufferSliceBufs.data()), sizeof(HcclAicpuLocBufLite) * descNum,
267 : RT_MEMCPY_HOST_TO_DEVICE);
268 :
269 2 : HrtMemcpy(
270 4 : reinterpret_cast<void*>(devBatchPutGetRemoteBufs->GetAddr()), devBatchPutGetRemoteBufs->GetSize(),
271 2 : static_cast<void*>(hostBatchPutGetRemoteBufferSliceBufs.data()), sizeof(HcclAicpuLocBufLite) * descNum,
272 : RT_MEMCPY_HOST_TO_DEVICE);
273 :
274 2 : return HCCL_SUCCESS;
275 2 : }
276 :
277 2 : std::vector<char> HcclOneSidedService::PackOpData(const CollAlgOpReq& req) const
278 : {
279 2 : std::vector<ModuleData> dataVec;
280 2 : dataVec.resize(AicpuResMgrType::__COUNT__);
281 :
282 2 : AicpuResMgrType resType = AicpuResMgrType::STREAM;
283 2 : OneSidedSetModuleDataName(dataVec[resType], "StreamManager");
284 2 : dataVec[resType].data = comm_->GetAicpuStreamManager().GetPackedData();
285 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
286 :
287 2 : resType = AicpuResMgrType::QUEUE_NOTIFY;
288 2 : OneSidedSetModuleDataName(dataVec[resType], "QueueNotifyManager");
289 2 : dataVec[resType].data = comm_->GetAicpuQueueNotifyManager().GetPackedData();
290 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
291 :
292 2 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
293 2 : OneSidedSetModuleDataName(dataVec[resType], "QueueWaitGroupCntNotifyManager");
294 2 : dataVec[resType].data = comm_->GetQueueWaitGroupCntNotifyManager().GetPackedData();
295 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
296 :
297 2 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
298 2 : OneSidedSetModuleDataName(dataVec[resType], "GetBcastPostCntNotifyManager");
299 2 : dataVec[resType].data = comm_->GetBcastPostCntNotifyManager().GetPackedData();
300 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
301 :
302 2 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
303 2 : OneSidedSetModuleDataName(dataVec[resType], "HostDeviceSyncNotifyManager");
304 2 : dataVec[resType].data = comm_->GetHostDeviceSyncNotifyManager().GetPackedData();
305 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
306 :
307 2 : resType = AicpuResMgrType::TRANSPORT;
308 2 : OneSidedSetModuleDataName(dataVec[resType], "MemTransportManager");
309 2 : auto op = comm_->GetCurrentCollOperator();
310 : // GetOpbasedPackedData由于单边通信隔离，会找不到Transport
311 2 : if (op->opMode == OpMode::OPBASE) { // 单算子模式
312 2 : dataVec[resType].data = comm_->GetMemTransportManager()->GetOneSidedPackedData();
313 : } else {
314 0 : THROW<InternalException>(StringFormat("opMode=%s failed", op->opMode.Describe().c_str()));
315 : }
316 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
317 :
318 2 : resType = AicpuResMgrType::ALG_TOPO;
319 2 : OneSidedSetModuleDataName(dataVec[resType], req.algName);
320 : AlgTopoPackageHelper algTopoHelper;
321 2 : dataVec[resType].data = algTopoHelper.GetPackedData(req.resReq.topoInfo);
322 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
323 :
324 2 : resType = AicpuResMgrType::CONNECTD_MGR;
325 2 : OneSidedSetModuleDataName(dataVec[resType], "ConnectedManager");
326 2 : dataVec[resType].data = comm_->GetRankGraph()->GetPackedData(req.resReq.levelRankPairs);
327 6 : HCCL_INFO("HcclOneSidedService::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
328 :
329 : AicpuResPackageHelper helper;
330 4 : return helper.GetPackedData(dataVec);
331 2 : }
332 :
333 2 : void HcclOneSidedService::FillOneSidedOperator(OpType type, RankId remoteRankId, const HcclOneSideOpDesc* desc) const
334 : {
335 2 : CollOpParams opParams;
336 :
337 2 : opParams.dataType = HcclDataTypeToDataType(desc->dataType);
338 2 : opParams.count = desc->count;
339 :
340 : // sendBuf/recvBuf当前不使用，等待后续扩展
341 2 : opParams.sendBuf = desc->localAddr;
342 2 : opParams.recvBuf = desc->remoteAddr;
343 :
344 2 : opParams.opType = type;
345 2 : opParams.dstRank = remoteRankId;
346 2 : std::string opTag = comm_->GetId();
347 :
348 6 : HCCL_INFO(
349 : "[HcclOneSidedService][FillOneSidedOperator] CovertToCurrentCollOperator opType[%s], dstRank[%u], opTag[%s]",
350 : opParams.opType.Describe().c_str(), opParams.dstRank, opTag.c_str());
351 2 : comm_->CovertToCurrentCollOperator(opTag, opParams, OpMode::OPBASE);
352 2 : }
353 :
354 2 : void HcclOneSidedService::AddPostToUserStream(const Stream& stream) const
355 : {
356 2 : auto postNotify = comm_->GetHostDeviceSyncNotifyManager().GetDeviceWaitNotify();
357 :
358 2 : postNotify->Post(stream);
359 2 : }
360 :
361 2 : void HcclOneSidedService::AddWaitToUserStream(const Stream& stream) const
362 : {
363 2 : auto waitNotify = comm_->GetHostDeviceSyncNotifyManager().GetHostWaitNotify();
364 :
365 2 : waitNotify->Wait(stream, 1000); // host 和 device sync流程，等待1000ms
366 2 : }
367 :
368 2 : void HcclOneSidedService::SetOneSidedKernelLaunchParam(HcclKernelLaunchParam& param, const DevBuffer* mem) const
369 : {
370 2 : CollOperator op = *comm_->GetCurrentCollOperator();
371 :
372 6 : HCCL_INFO("[HcclOneSidedService][SetOneSidedKernelLaunchParam] op.opType[%s]", op.opType.Describe().c_str());
373 2 : param.kernel.comm.idIndex = comm_->GetIdIndex();
374 2 : param.kernel.comm.myRank = comm_->GetMyRank();
375 2 : param.kernel.comm.rankSize = comm_->GetRankSize();
376 2 : param.kernel.comm.devType = comm_->GetDevType();
377 2 : param.kernel.comm.devPhyId = comm_->GetDevicePhyId();
378 2 : param.kernel.comm.opCounterAddr = static_cast<u64>(counterBuf->GetAddr());
379 2 : auto ret = strcpy_s(param.kernel.comm.commId, sizeof(param.kernel.comm.commId), comm_->GetId().data());
380 2 : if (ret != EOK) {
381 0 : THROW<InternalException>(
382 0 : StringFormat("HcclOneSidedService::SetOneSidedKernelLaunchParam, strcpy_s commId failed! ret[%d]", ret));
383 : }
384 :
385 2 : param.kernel.oneSidedComm = true;
386 :
387 2 : param.kernel.op.algOperator.opMode = op.opMode;
388 2 : param.kernel.op.algOperator.opType = op.opType;
389 :
390 2 : param.kernel.binaryResAddr = mem->GetAddr();
391 2 : param.kernel.binaryResSize = mem->GetSize();
392 :
393 2 : param.kernel.op.sendRecvRemoteRank = op.sendRecvRemoteRank;
394 :
395 2 : param.kernel.kfcControlTransferH2DParams = comm_->GetKfcControlTransferH2D().GetCommunicateParams();
396 2 : param.kernel.kfcControlTransferD2HParams = comm_->GetKfcStatusTransferD2H().GetCommunicateParams();
397 2 : }
398 :
399 2 : void HcclOneSidedService::OneSidedAicpuKernelLaunch(HcclKernelLaunchParam& param, Stream& stream) const
400 : {
401 2 : const aclrtFuncHandle funcHandle = comm_->GetAicpuKernelFuncHandle(param.kernelName);
402 2 : constexpr u32 numBlocks = 1;
403 : aclrtLaunchKernelCfg cfg;
404 : aclrtLaunchKernelAttr attr;
405 2 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
406 2 : auto timeoutCheck = EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
407 : // aicpu kernal超时时间: X+30s
408 2 : attr.value.timeout = static_cast<u16>((timeoutCheck == 0) ? timeoutCheck : (timeoutCheck + 30));
409 2 : cfg.numAttrs = 1;
410 2 : cfg.attrs = &attr;
411 2 : AddPostToUserStream(stream);
412 6 : HCCL_INFO(
413 : "[HcclOneSidedService::AicpuKernelLaunch] param.soName: %s, param.kernelName: %s", param.soName,
414 : param.kernelName);
415 2 : HrtAicpuLaunchKernelWithHostArgs(
416 2 : funcHandle, numBlocks, comm_->GetAicpuStreamManager().GetFreeStream()->GetPtr(), &cfg, ¶m.kernel,
417 : sizeof(HcclKernelParamLite));
418 6 : HCCL_INFO(
419 : "[HcclOneSidedService][AicpuKernelLaunch] param.kernel.algName: %s HrtAicpuLaunchKernelWithHostArgs end!",
420 : param.kernel.algName);
421 2 : AddWaitToUserStream(stream);
422 2 : }
423 :
424 2 : DevBuffer* HcclOneSidedService::PackResToKernelLanuch(CollAlgOpReq& opReq)
425 : {
426 2 : auto it = OneSidedLoadMap.find(opReq.algName);
427 2 : if (it != OneSidedLoadMap.end()) { // 已经向Device Mem写过资源
428 0 : HCCL_INFO("[OpBasedCollProcess] tag[%s] devMem has been allocated, reuse it", opReq.algName.c_str());
429 0 : return it->second.get();
430 : }
431 :
432 6 : HCCL_INFO("[HcclOneSidedService][PackResToKernelLanuch], PackOpData start");
433 : // 打包单边通信资源信息到device
434 2 : auto buffer = PackOpData(opReq);
435 2 : shared_ptr<DevBuffer> devMem = make_shared<DevBuffer>(buffer.size()); // 申请device内存
436 :
437 6 : HCCL_INFO("[HcclOneSidedService][PackResToKernelLanuch], HrtMemSyncCopy start");
438 4 : HrtMemcpy(
439 4 : reinterpret_cast<void*>(devMem->GetAddr()), devMem->GetSize(), buffer.data(), buffer.size(),
440 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到device内存
441 6 : HCCL_INFO(
442 : "HcclOneSidedService::BatchGet PackOpData: PackedData %s", Bytes2hex(buffer.data(), buffer.size()).c_str());
443 2 : OneSidedLoadMap.insert(make_pair(opReq.algName, devMem));
444 :
445 2 : return devMem.get();
446 2 : }
447 :
448 2 : HcclResult HcclOneSidedService::BatchOpKernelLaunch(
449 : OpType opType, RankId remoteRankId, const HcclOneSideOpDesc* desc, u32 descNum, shared_ptr<Stream> stream)
450 : {
451 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] start");
452 2 : comm_->GetAicpuStreamManager().AllocFreeStream();
453 6 : HCCL_INFO("[HcclOneSidedService][AllocStreams] start");
454 2 : comm_->GetAicpuStreamManager().AllocStreams(1);
455 :
456 2 : CHK_PTR_NULL(desc);
457 6 : HCCL_INFO(
458 : "[HcclOneSidedService][BatchOpKernelLaunch] desc: localAddr:[%p],remoteAddr:[%p],count:[%llu],dataType:[%d]",
459 : desc->localAddr, desc->remoteAddr, desc->count, desc->dataType);
460 :
461 2 : CollAlgOpReq opReq;
462 2 : opReq.algName = OpTypeToString(opType);
463 2 : opReq.resReq.levelRankPairs.push_back(make_pair(0, remoteRankId));
464 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] FillOneSidedOperator start");
465 : // 填充通信算子信息
466 2 : FillOneSidedOperator(opType, remoteRankId, desc);
467 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] PackResToKernelLanuch start");
468 : // 打包device展开资源信息
469 2 : DevBuffer* devMem = PackResToKernelLanuch(opReq);
470 : // 组kernelLaunch参数
471 2 : HcclKernelLaunchParam param;
472 :
473 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] SetOneSidedKernelLaunchParam start");
474 : // 构造单边通信公共参数
475 2 : SetOneSidedKernelLaunchParam(param, devMem);
476 :
477 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
478 2 : auto it = oneSidedConns_.find(remoteRankId);
479 2 : if (it == oneSidedConns_.end()) {
480 0 : HCCL_ERROR("[HcclMemCommunication][BatchGet] Can't find oneSidedConn by remoteRank %u", remoteRankId);
481 0 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
482 : }
483 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] BatchPutGetDevBufs start");
484 2 : CHK_RET(BatchPutGetDevBufs(desc, descNum, it->second));
485 2 : oneSidedConnslock.unlock();
486 :
487 2 : param.kernel.op.batchPutGetDescNum = descNum;
488 2 : param.kernel.op.batchPutGetLocalAddr = reinterpret_cast<void*>(devBatchPutGetLocalBufs.get()->GetAddr());
489 2 : param.kernel.op.batchPutGetRemoteAddr = reinterpret_cast<void*>(devBatchPutGetRemoteBufs.get()->GetAddr());
490 2 : auto ret = strcpy_s(param.kernel.tagKey, sizeof(param.kernel.tagKey), opReq.algName.c_str());
491 2 : if (ret != EOK) {
492 0 : THROW<InternalException>(
493 0 : StringFormat("[HcclOneSidedService][BatchOpKernelLaunch], strcpy_s opReq.algName failed! ret[%d]", ret));
494 : }
495 :
496 6 : HCCL_INFO("[HcclOneSidedService][BatchOpKernelLaunch] OneSidedAicpuKernelLaunch start");
497 : // 启动kernel
498 2 : OneSidedAicpuKernelLaunch(param, *stream);
499 2 : return HCCL_SUCCESS;
500 2 : }
501 :
502 : HcclResult
503 773 : HcclOneSidedService::BatchPut(RankId remoteRankId, const HcclOneSideOpDesc* desc, u32 descNum, const rtStream_t stream)
504 : {
505 2325 : HCCL_INFO("[HcclOneSidedService][BatchPut] start");
506 766 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
507 758 : auto it = oneSidedConns_.find(remoteRankId);
508 647 : if (it == oneSidedConns_.end()) {
509 2182 : HCCL_ERROR("[HcclMemCommunication][BatchPut] Can't find oneSidedConn by remoteRank %u", remoteRankId);
510 782 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
511 : }
512 1 : oneSidedConnslock.unlock();
513 :
514 3 : HCCL_INFO("[HcclOneSidedService][BatchPut] BatchOpKernelLaunch start");
515 1 : CHK_RET(BatchOpKernelLaunch(OpType::BATCHPUT, remoteRankId, desc, descNum, std::make_shared<Stream>(stream)));
516 :
517 3 : HCCL_INFO("[HcclOneSidedService][BatchPut] end");
518 1 : return HCCL_SUCCESS;
519 800 : }
520 :
521 : HcclResult
522 2 : HcclOneSidedService::BatchGet(RankId remoteRankId, const HcclOneSideOpDesc* desc, u32 descNum, const rtStream_t stream)
523 : {
524 6 : HCCL_INFO("[HcclOneSidedService][BatchGet] start");
525 2 : std::shared_lock oneSidedConnslock(oneSidedConnsMutex_);
526 2 : auto it = oneSidedConns_.find(remoteRankId);
527 2 : if (it == oneSidedConns_.end()) {
528 3 : HCCL_ERROR("[HcclMemCommunication][BatchGet] Can't find oneSidedConn by remoteRank %u", remoteRankId);
529 1 : throw out_of_range("Can't find oneSidedConn by remoteRank.");
530 : }
531 1 : oneSidedConnslock.unlock();
532 :
533 3 : HCCL_INFO("[HcclOneSidedService][BatchGet] BatchOpKernelLaunch start");
534 1 : CHK_RET(BatchOpKernelLaunch(OpType::BATCHGET, remoteRankId, desc, descNum, std::make_shared<Stream>(stream)));
535 :
536 3 : HCCL_INFO("[HcclOneSidedService][BatchGet] end");
537 1 : return HCCL_SUCCESS;
538 2 : }
539 :
540 58 : void HcclOneSidedService::AddOpCounterMems()
541 : {
542 174 : HCCL_INFO("[HcclOneSidedService::%s] start.", __func__);
543 :
544 58 : constexpr u64 FOUR_BYTES = 4;
545 58 : u64 size = FOUR_BYTES * 3; // 第一个四字节用于计数加1, 后面两个四字节分别保存headCounter和tailCounter
546 58 : counterBuf = std::make_shared<DevBuffer>(size);
547 :
548 : // 初始化第一个四字节置1, 用于计数加1, reduce task add 1
549 58 : u64 srcSize = FOUR_BYTES;
550 58 : float srcValue = 1;
551 58 : void* srcAddr = reinterpret_cast<void*>(counterBuf->GetAddr());
552 58 : HrtMemcpy(srcAddr, srcSize, &srcValue, srcSize, RT_MEMCPY_HOST_TO_DEVICE);
553 :
554 : // 初始化后面两个四字节置0
555 58 : u64 countMemSize = srcSize;
556 58 : float startValue = 0; // value为0表示从0开始计数
557 58 : void* headCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize);
558 58 : void* tailCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize * 2);
559 58 : HrtMemcpy(headCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
560 58 : HrtMemcpy(tailCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
561 :
562 174 : HCCL_INFO(
563 : "[HcclOneSidedService::%s] end, counterBuf[%llu] srcAddr[%p] headCountAddr[%p] tailCountAddr[%p].", __func__,
564 : counterBuf->GetAddr(), srcAddr, headCountAddr, tailCountAddr);
565 58 : }
566 :
567 0 : DevBuffer* HcclOneSidedService::GetOpCounterBuf() { return counterBuf.get(); }
568 : } // namespace Hccl
|
