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 : #include "channel_manager.h"
11 : #include "adapter_rts_common.h"
12 : #include "log.h"
13 : #include "comm_configer.h"
14 : #include "launch_aicpu.h"
15 : #include "comm_engine_utils.h"
16 : #include <unordered_set>
17 : #include <string>
18 : #include "adapter_prof.h"
19 : #include "hcom_host_profiling.h"
20 :
21 : namespace hccl {
22 :
23 : constexpr u32 RDMA_NOTIFY_MIN_NUM = 3;
24 : constexpr u32 NOTIFY_NUM_MAX = 64; // HcclChannelDesc 中 notifynum 的默认限制最大为64
25 :
26 261 : HcclResult ChannelManager::Init(aclrtBinHandle binHandle, u32 userRank, const ManagerCallbacks& callbacks)
27 : {
28 261 : binHandle_ = binHandle;
29 261 : userRank_ = userRank;
30 261 : callbacks_ = callbacks;
31 261 : return HCCL_SUCCESS;
32 : }
33 :
34 404 : HcclResult ChannelManager::SetChannelCallbacks(const ChannelManagerCallbacks& channelCallbacks)
35 : {
36 404 : channelCallbacks_ = channelCallbacks;
37 404 : rankInfoList_ = channelCallbacks_.getRankLists();
38 404 : return HCCL_SUCCESS;
39 : }
40 :
41 6 : std::vector<HcclMemHandle> ChannelManager::CollectMemHandles(const std::vector<HcclChannelDesc>& descs) const
42 : {
43 6 : std::vector<HcclMemHandle> handles;
44 6 : std::unordered_set<HcclMemHandle> seen;
45 15 : for (const auto& desc : descs) {
46 18 : for (uint32_t i = 0; i < desc.memHandleNum; ++i) {
47 9 : if (desc.memHandles != nullptr && seen.insert(desc.memHandles[i]).second) {
48 7 : handles.push_back(desc.memHandles[i]);
49 : }
50 : }
51 : }
52 6 : return handles;
53 6 : }
54 :
55 7 : HcclResult ChannelManager::CheckChannelParam(CommEngine engine, const HcclChannelDesc* channelDesc, uint32_t descNum)
56 : {
57 7 : std::unordered_set<HcclChannelDesc, std::hash<HcclChannelDesc>, HcclChannelDescEqual> descSet;
58 :
59 10 : for (uint32_t descIdx = 0; descIdx < descNum; ++descIdx) {
60 : // 检查notifyNum
61 9 : CHK_PRT_RET(
62 : channelDesc[descIdx].notifyNum > NOTIFY_NUM_MAX,
63 : HCCL_ERROR(
64 : "[%s]Channeldesc[%u] invalid notifyNum, notifyNum[%u], max notify num[%u]", __func__, descIdx,
65 : channelDesc[descIdx].notifyNum, NOTIFY_NUM_MAX),
66 : HCCL_E_PARA);
67 : // 检查HcclChannelDesc是否有重复元素
68 7 : CHK_PRT_RET(
69 : descSet.find(channelDesc[descIdx]) != descSet.end(),
70 : HCCL_ERROR("[%s]Duplicate item found in hcclchanneldesc.", __func__), HCCL_E_PARA);
71 6 : descSet.insert(channelDesc[descIdx]);
72 : // 检查RemoteRank有效性
73 6 : CHK_PRT_RET(
74 : channelDesc[descIdx].remoteRank == userRank_,
75 : HCCL_ERROR("[%s]Local rank found in channeldesc, userRank_ = %u.", __func__, userRank_), HCCL_E_PARA);
76 : // 检查是否有不支持协议
77 5 : if (engine == COMM_ENGINE_AIV) {
78 0 : CHK_PRT_RET(
79 : channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS
80 : && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS_ONLY
81 : && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_SIO,
82 : HCCL_ERROR(
83 : "[%s]AIV engine Unsupported protocol[%d] found in channeldesc, protocol: %d.", __func__, descIdx,
84 : channelDesc[descIdx].channelProtocol),
85 : HCCL_E_PARA);
86 : } else {
87 5 : CHK_PRT_RET(
88 : channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS
89 : && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_ROCE
90 : && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_HCCS_ONLY
91 : && channelDesc[descIdx].channelProtocol != COMM_PROTOCOL_SIO,
92 : HCCL_ERROR(
93 : "[%s]Unsupported protocol[%d] found in channeldesc, protocol: %d.", __func__, descIdx,
94 : channelDesc[descIdx].channelProtocol),
95 : HCCL_E_PARA);
96 : }
97 :
98 : // 检查engine支持情况
99 4 : if (engine != COMM_ENGINE_CPU && engine != COMM_ENGINE_CPU_TS && engine != COMM_ENGINE_AICPU
100 1 : && engine != COMM_ENGINE_AICPU_TS && engine != COMM_ENGINE_AIV) {
101 1 : HCCL_ERROR(
102 : "[%s]Unsupported engine for channel, engine: %s.", __func__,
103 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
104 1 : return HCCL_E_PARA;
105 : }
106 : }
107 1 : return HCCL_SUCCESS;
108 7 : }
109 :
110 28 : static std::string BuildChannelKey(const std::string& tag, CommEngine engine, const HcclChannelDesc& channelDesc)
111 : {
112 56 : std::string key = tag + ":" + std::to_string(engine) + ":" + std::to_string(channelDesc.remoteRank) + ":"
113 84 : + std::to_string(channelDesc.channelProtocol);
114 : // 将 memHandles 纳入 key，使得上层更换交换内存时会强制新建 channel，避免复用旧 channel 导致远端地址错配
115 28 : if (channelDesc.memHandleNum > 0 && channelDesc.memHandles != nullptr) {
116 22 : for (uint32_t i = 0; i < channelDesc.memHandleNum; i++) {
117 13 : key += ":" + std::to_string(reinterpret_cast<uintptr_t>(channelDesc.memHandles[i]));
118 : }
119 9 : } else {
120 19 : key += ":0";
121 : }
122 28 : return key;
123 0 : }
124 :
125 18 : HcclResult ChannelManager::RegisterHandle(
126 : const std::string& tag, CommEngine engine, const HcclChannelDesc& channelDesc, ChannelHandle channelHandle)
127 : {
128 18 : std::string channelKey = BuildChannelKey(tag, engine, channelDesc);
129 :
130 18 : CHK_PRT_RET(
131 : (channelHandleMap_.find(channelKey) != channelHandleMap_.end()),
132 : HCCL_ERROR(
133 : "[%s]Channel already exists, tag[%s], engine[%s], remoteRank[%d], channelProtocol[%d].", __func__,
134 : tag.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelDesc.remoteRank,
135 : channelDesc.channelProtocol),
136 : HCCL_E_PARA);
137 17 : channelHandleMap_[channelKey] = channelHandle;
138 17 : keyMap_[channelHandle] = channelKey;
139 17 : engineMap_[channelHandle] = engine;
140 17 : HCCL_INFO("[%s]Register channel handle[%llu], channelKey[%s]", __func__, channelHandle, channelKey.c_str());
141 17 : return HCCL_SUCCESS;
142 18 : }
143 :
144 8 : HcclResult ChannelManager::PrepareHandleArray(
145 : const std::string& tag, CommEngine engine, const HcclChannelDesc* channelDesc, uint32_t descNum,
146 : ChannelHandle* channelHandleArray, std::vector<HcclChannelDesc>& needCreateDescs,
147 : std::vector<uint32_t>& needCreateIndices)
148 : {
149 8 : needCreateDescs.clear();
150 8 : needCreateIndices.clear();
151 :
152 18 : for (uint32_t descIdx = 0; descIdx < descNum; descIdx++) {
153 : // 组合channelKey
154 10 : std::string channelKey = BuildChannelKey(tag, engine, channelDesc[descIdx]);
155 10 : if (channelHandleMap_.find(channelKey) != channelHandleMap_.end()) {
156 4 : channelHandleArray[descIdx] = channelHandleMap_[channelKey];
157 4 : continue;
158 : }
159 6 : channelHandleArray[descIdx] = 0;
160 6 : needCreateDescs.push_back(channelDesc[descIdx]);
161 6 : needCreateIndices.push_back(descIdx);
162 10 : }
163 :
164 8 : return HCCL_SUCCESS;
165 : }
166 :
167 4 : HcclResult ChannelManager::IsChannelExist(ChannelHandle channel)
168 : {
169 4 : CHK_PRT_RET(
170 : (keyMap_.find(channel) == keyMap_.end()), HCCL_ERROR("[%s]ChannelHandle is not exist.", __func__), HCCL_E_PARA);
171 1 : HCCL_INFO(
172 : "[%s]ChannelHandle exist, ChannelHandle[%llu], channelKey[%s]", __func__, channel, keyMap_[channel].c_str());
173 1 : return HCCL_SUCCESS;
174 : }
175 :
176 6 : HcclResult ChannelManager::UnregisterHandle(ChannelHandle channel)
177 : {
178 6 : CHK_PRT_RET(
179 : (keyMap_.find(channel) == keyMap_.end()), HCCL_ERROR("[%s]ChannelHandle is not exist.", __func__), HCCL_E_PARA);
180 :
181 4 : channelHandleMap_.erase(keyMap_[channel]);
182 4 : keyMap_.erase(channel);
183 4 : if (engineMap_[channel] == COMM_ENGINE_AICPU || engineMap_[channel] == COMM_ENGINE_AICPU_TS) {
184 0 : channelD2HMap_.erase(channel);
185 : }
186 4 : engineMap_.erase(channel);
187 :
188 4 : HCCL_INFO("[%s]Unregister channel handle success.", __func__);
189 4 : return HCCL_SUCCESS;
190 : }
191 :
192 6 : HcclResult ChannelManager::RegisterHandleHDPair(ChannelHandle deviceChannelHandle, ChannelHandle hostChannelHandle)
193 : {
194 6 : CHK_PRT_RET(
195 : (deviceChannelHandle == 0 || hostChannelHandle == 0), HCCL_ERROR("[%s]ChannelHandle is 0.", __func__),
196 : HCCL_E_PARA);
197 4 : CHK_PRT_RET(
198 : (channelD2HMap_.find(deviceChannelHandle) != channelD2HMap_.end()),
199 : HCCL_ERROR("[%s]deviceChannelHandle has existed in channelD2HMap_.", __func__), HCCL_E_PARA);
200 :
201 3 : channelD2HMap_[deviceChannelHandle] = hostChannelHandle;
202 3 : return HCCL_SUCCESS;
203 : }
204 :
205 3 : HcclResult ChannelManager::GetHostChannel(ChannelHandle channel, ChannelHandle& hostChannel)
206 : {
207 3 : if (engineMap_[channel] == COMM_ENGINE_AICPU || engineMap_[channel] == COMM_ENGINE_AICPU_TS) {
208 2 : CHK_PRT_RET(
209 : (channelD2HMap_.find(channel) == channelD2HMap_.end()),
210 : HCCL_ERROR("[%s]device channel handle has not existed in channelD2HMap_.", __func__), HCCL_E_PARA);
211 1 : hostChannel = channelD2HMap_[channel];
212 : } else {
213 1 : hostChannel = channel;
214 : }
215 2 : return HCCL_SUCCESS;
216 : }
217 :
218 4 : void ChannelManager::ClearOpTransportResponseLinks(OpCommTransport& opTransportResponse)
219 : {
220 8 : for (auto& levelNSubCommTransport : opTransportResponse) {
221 8 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
222 4 : u32 size = singleSubCommTransport.transportRequests.size();
223 4 : singleSubCommTransport.links.resize(size, nullptr);
224 4 : singleSubCommTransport.status.resize(size, TransportStatus::INIT);
225 4 : HCCL_INFO("[%s] size[%u], linksSize[%zu]", __func__, size, singleSubCommTransport.links.size());
226 : }
227 : }
228 4 : }
229 :
230 3 : HcclResult ChannelManager::CheckNotifyOrQPMaxNum(u64& existNum, const u64& MaxNum, const bool& isNotifyRes)
231 : {
232 3 : std::string resType = isNotifyRes ? "Notify" : "QP";
233 3 : if (existNum + 1 > MaxNum) {
234 1 : HCCL_ERROR(
235 : "[%s]%s resources are insufficient, existNum[%llu], MaxNum is [%llu]", __func__, resType.c_str(), existNum,
236 : MaxNum);
237 1 : return HCCL_E_INTERNAL;
238 : }
239 2 : HCCL_DEBUG(
240 : "[%s]%s resources are sufficient, existNum[%llu], MaxNum is [%llu]", __func__, resType.c_str(), existNum,
241 : MaxNum);
242 2 : return HCCL_SUCCESS;
243 3 : }
244 :
245 0 : HcclResult ChannelManager::CreateWorkSpace(u64 size, DeviceMem& buffer) const
246 : {
247 0 : CHK_PRT_RET(
248 : size == 0, HCCL_INFO("[Create][WorkSpace]work space size is zero. not need to malloc memory"), HCCL_SUCCESS);
249 :
250 : CHK_PRT_RET(
251 : (size > ULONG_MAX), HCCL_ERROR("[Create][WorkSpace]work space size is greater than %llu", ULONG_MAX),
252 : HCCL_E_PARA);
253 :
254 0 : u64 memSize = size;
255 0 : buffer = DeviceMem::alloc(memSize);
256 0 : CHK_PRT_RET(
257 : size > 0 && !buffer,
258 : HCCL_ERROR(
259 : "[Create][WorkSpace]Create work space size[%llu] fail,"
260 : "please check workspace size.",
261 : size),
262 : HCCL_E_PTR);
263 0 : CHK_RET(hrtMemSet(buffer.ptr(), size, size));
264 0 : return HCCL_SUCCESS;
265 : }
266 :
267 0 : HcclResult ChannelManager::AllocAndClearHostMem(u64 size, std::shared_ptr<HostMem>& bufferPtr) const
268 : {
269 0 : CHK_PRT_RET(
270 : size == 0,
271 : HCCL_INFO("[ChannelManager][AllocAndClearHostMem] host memory size is zero. not need to malloc memory"),
272 : HCCL_SUCCESS);
273 :
274 : CHK_PRT_RET(
275 : (size > ULONG_MAX),
276 : HCCL_ERROR("[ChannelManager][AllocAndClearHostMem] host memory size is greater than %llu", ULONG_MAX),
277 : HCCL_E_PARA);
278 :
279 0 : HostMem tmpBuffer = HostMem::alloc(size);
280 0 : EXCEPTION_CATCH((bufferPtr = std::make_shared<HostMem>(std::move(tmpBuffer))), return HCCL_E_PTR);
281 :
282 0 : CHK_PRT_RET(
283 : size > 0 && !bufferPtr.get()->ptr(),
284 : HCCL_ERROR(
285 : "[ChannelManager][AllocAndClearHostMem]host memory space size[%llu] fail,"
286 : "please check workspace size.",
287 : size),
288 : HCCL_E_PTR);
289 0 : CHK_SAFETY_FUNC_RET(memset_s(bufferPtr.get()->ptr(), size, 0, size));
290 0 : return HCCL_SUCCESS;
291 0 : }
292 :
293 : template <typename T>
294 0 : HcclResult ChannelManager::CopyVectorToDeviceMem(const u64 len, DeviceMem& dstDeviceMem, const std::vector<T>& srcVec)
295 : {
296 0 : CHK_PRT_RET(
297 : len == 0, HCCL_INFO("[ChannelManager][CopyVectorToDeviceMem] space size is zero. not need to malloc memory"),
298 : HCCL_SUCCESS);
299 :
300 : CHK_PRT_RET(
301 : (len > ULONG_MAX),
302 : HCCL_ERROR("[ChannelManager][CopyVectorToDeviceMem] space size is greater than %llu", ULONG_MAX), HCCL_E_PARA);
303 :
304 0 : CHK_RET(CreateWorkSpace(len, dstDeviceMem));
305 0 : std::shared_ptr<HostMem> srcHostMem;
306 0 : CHK_RET(AllocAndClearHostMem(len, srcHostMem));
307 0 : std::copy(srcVec.begin(), srcVec.end(), static_cast<T*>(srcHostMem.get()->ptr()));
308 0 : CHK_RET(hrtMemSyncCopy(
309 : dstDeviceMem.ptr(), len, srcHostMem.get()->ptr(), len, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
310 0 : return HCCL_SUCCESS;
311 0 : }
312 :
313 3 : OpCommTransport ChannelManager::BuildChannelRequests(const std::vector<HcclChannelDesc>& descs)
314 : {
315 3 : OpCommTransport opCommTransport;
316 3 : LevelNSubCommTransport level0Transport;
317 3 : SingleSubCommTransport commTransport;
318 :
319 7 : for (auto desc : descs) {
320 4 : TransportRequest tmpTransport;
321 4 : tmpTransport.isValid = true;
322 4 : tmpTransport.localUserRank = userRank_;
323 4 : tmpTransport.remoteUserRank = desc.remoteRank;
324 4 : tmpTransport.notifyNum = desc.notifyNum;
325 4 : tmpTransport.inputMemType = TransportMemType::CCL_INPUT;
326 4 : tmpTransport.outputMemType = TransportMemType::CCL_OUTPUT;
327 4 : tmpTransport.isUsedRdma = (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_ROCE);
328 4 : TransportLinkType linkType = TransportLinkType::RESERVED;
329 4 : if (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_HCCS_ONLY) {
330 0 : linkType = TransportLinkType::HCCS;
331 4 : } else if (desc.channelProtocol == CommProtocol::COMM_PROTOCOL_SIO) {
332 1 : linkType = TransportLinkType::SIO;
333 : }
334 4 : tmpTransport.linkType = linkType;
335 4 : commTransport.transportRequests.push_back(tmpTransport);
336 : }
337 :
338 3 : level0Transport.push_back(commTransport);
339 3 : opCommTransport.push_back(level0Transport);
340 3 : ClearOpTransportResponseLinks(opCommTransport);
341 :
342 3 : return opCommTransport;
343 3 : }
344 :
345 0 : HcclResult ChannelManager::ParseChannelRemoteDataToMem(
346 : const OpCommTransport& opTransportResponse, HcclIndOpChannelRemoteResV3& channelParam)
347 : {
348 0 : uint32_t level0 = 0;
349 0 : auto& singleSubCommTransport = opTransportResponse[level0][level0];
350 0 : CHK_PRT_RET(
351 : channelParam.listNum == 0, HCCL_ERROR("[%s]invalid listNum, listNum[%u]", __func__, channelParam.listNum),
352 : HCCL_E_PARA);
353 0 : CHK_PRT_RET(
354 : (channelParam.listNum != singleSubCommTransport.links.size()),
355 : HCCL_ERROR(
356 : "[%s]invalid listNum, listNum[%u] but links size is [%zu]", __func__, channelParam.listNum,
357 : singleSubCommTransport.links.size()),
358 : HCCL_E_PARA);
359 : // 分配 HcclIndOpChannelRemoteResV2 内存，需要手动释放
360 : channelParam.remoteResV2
361 0 : = static_cast<HcclIndOpChannelRemoteResV2*>(malloc(channelParam.listNum * sizeof(HcclIndOpChannelRemoteResV2)));
362 0 : CHK_PRT_RET(
363 : channelParam.remoteResV2 == nullptr, HCCL_ERROR("[%s]channelParam.remoteResV2 is null.", __func__),
364 : HCCL_E_MEMORY);
365 0 : u32 linkIdx = 0;
366 0 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
367 0 : auto& tempLink = singleSubCommTransport.links[linkIdx];
368 0 : channelParam.remoteResV2[linkIdx].remoteWorldRank = rankInfoList_[transportRequest.remoteUserRank].worldRank;
369 0 : channelParam.remoteResV2[linkIdx].remoteRank = transportRequest.remoteUserRank;
370 : // transport信息保存（notify、qp）
371 0 : if (!transportRequest.isUsedRdma) {
372 : // sdma -> P2P
373 0 : CHK_RET(BuildOpRemoteChannelP2pResParam(tempLink, channelParam.remoteResV2[linkIdx]));
374 0 : channelParam.remoteResV2[linkIdx].channelP2p.qos = hcclQos_;
375 0 : HCCL_INFO(
376 : "[ChannelManager] [ParseChannelRemoteDataToMem] hcclQos[%u]",
377 : channelParam.remoteResV2[linkIdx].channelP2p.qos);
378 : } else {
379 : // rdma -> roce
380 0 : CHK_RET(BuildOpRemoteChannelRoceResParam(tempLink, channelParam.remoteResV2[linkIdx]));
381 : }
382 0 : linkIdx++;
383 : }
384 0 : return HCCL_SUCCESS;
385 : }
386 :
387 0 : HcclResult ChannelManager::BuildOpRemoteChannelP2pResParam(const LINK& link, HcclIndOpChannelRemoteResV2& remoteRes)
388 : {
389 0 : remoteRes.isUsedRdma = false;
390 0 : HcclChannelP2p& linkp2p = remoteRes.channelP2p;
391 : // remoteMem, 独立算子localmem是否需要传待确认
392 0 : void* bufferPtr = nullptr;
393 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &bufferPtr));
394 0 : linkp2p.remoteHcclbuffer.addr = reinterpret_cast<void*>(bufferPtr);
395 : u64 remotebufferSize;
396 0 : CHK_RET(link->GetRemoteMemSize(UserMemType::INPUT_MEM, remotebufferSize));
397 0 : linkp2p.remoteHcclbuffer.size = remotebufferSize;
398 : // 独立算子远端用户内存，linkp2p.remoteUserMem需要手动释放内存
399 0 : CHK_RET(link->GetIndOpRemoteMem(&linkp2p.remoteUserMem, &linkp2p.remoteUserMemCount));
400 0 : HCCL_DEBUG("[%s] finish set remoteMem info", __func__);
401 :
402 : // localnotify & remotenotify
403 0 : u64 notifyNum = 0;
404 0 : std::vector<HcclSignalInfo> locIpcSignals;
405 0 : std::vector<HcclSignalInfo> rmtIpcSignals;
406 0 : CHK_RET(link->GetLocalNotify(locIpcSignals));
407 0 : CHK_RET(link->GetRemoteNotify(rmtIpcSignals));
408 :
409 0 : for (size_t i = 0; i < locIpcSignals.size(); i++) {
410 0 : linkp2p.localIpcSignal[notifyNum] = locIpcSignals[i];
411 0 : linkp2p.remoteIpcSignal[notifyNum] = rmtIpcSignals[i];
412 0 : notifyNum++;
413 : }
414 0 : remoteRes.p2pNotifyNum = link->GetNotifyNum();
415 0 : HCCL_DEBUG(
416 : "[%s] finish set localnotify & remotenotify info, notifyNum[%llu], p2pNotifyNum[%llu]", __func__, notifyNum,
417 : remoteRes.p2pNotifyNum);
418 : // transportAttr
419 0 : CHK_RET(link->GetTransportAttr(linkp2p.transportAttr));
420 0 : HCCL_DEBUG("[%s] finish set RemoteChannelP2pResParam info", __func__);
421 0 : return HCCL_SUCCESS;
422 0 : }
423 :
424 0 : HcclResult ChannelManager::BuildOpRemoteChannelRoceResParam(const LINK& link, HcclIndOpChannelRemoteResV2& remoteRes)
425 : {
426 0 : remoteRes.isUsedRdma = true;
427 0 : HcclChannelRoce& linkRoce = remoteRes.channelRoce;
428 : // 填充localMem信息到linkRoce中
429 0 : CHK_RET(link->GetLocalMemDetails(UserMemType::INPUT_MEM, linkRoce.localHcclbuffer));
430 : // 填充remoteMem信息到linkRoce中
431 0 : void* bufferPtr = nullptr;
432 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &bufferPtr));
433 0 : linkRoce.remoteHcclbuffer.addr = reinterpret_cast<u64>(bufferPtr);
434 0 : CHK_RET(link->GetRemoteMemKey(UserMemType::INPUT_MEM, &(linkRoce.remoteHcclbuffer.key)));
435 0 : CHK_RET(link->GetRemoteMemSize(UserMemType::INPUT_MEM, linkRoce.remoteHcclbuffer.size));
436 : // 独立算子远端用户内存，linkRoce.remoteUserHostMem和remoteUserDeviceMem需要手动释放内存
437 0 : CHK_RET(link->GetIndOpRemoteMemDetails(
438 : &linkRoce.remoteUserHostMem, &linkRoce.remoteUserHostMemCount, HcclMemType::HCCL_MEM_TYPE_HOST));
439 0 : CHK_RET(link->GetIndOpRemoteMemDetails(
440 : &linkRoce.remoteUserDeviceMem, &linkRoce.remoteUserHostMemCount, HcclMemType::HCCL_MEM_TYPE_DEVICE));
441 0 : HCCL_DEBUG("[%s] finish set remoteMem info", __func__);
442 :
443 : // 填充notifyValue和notifyValueKey信息到linkRoce中
444 0 : std::vector<AddrKey> notifyValueAddrKey;
445 0 : CHK_RET(link->GetLocalNotifyValueAddrKey(notifyValueAddrKey));
446 0 : linkRoce.notifyValue = notifyValueAddrKey[0].addr;
447 0 : linkRoce.notifyValueKey = notifyValueAddrKey[0].key;
448 :
449 : // 填充QP信息到linkRoce中
450 0 : std::vector<HcclQpInfoV2> aiQpInfos;
451 0 : CHK_RET(link->GetAiQpInfo(aiQpInfos));
452 0 : u32 qpNum = aiQpInfos.size();
453 0 : if (qpNum > RDMA_QP_MAX_NUM || qpNum < 1) {
454 0 : return HCCL_E_INTERNAL;
455 : }
456 0 : std::copy_n(aiQpInfos.begin(), qpNum, linkRoce.QpInfo);
457 0 : linkRoce.qpsPerConnection = qpNum - static_cast<u32>(qpNum > 1); // 多QP数量或单QP模式
458 :
459 : // 填充localNotify和remoteNotify信息到linkRoce中
460 0 : std::vector<AddrKey> notifyAddrKey;
461 0 : std::vector<HcclSignalInfo> signalInfos;
462 0 : CHK_RET(link->GetLocalRdmaNotify(signalInfos));
463 0 : CHK_RET(link->GetRemoteRdmaNotifyAddrKey(notifyAddrKey));
464 0 : if ((signalInfos.size() != notifyAddrKey.size()) || (signalInfos.size() < RDMA_NOTIFY_MIN_NUM)
465 0 : || (signalInfos.size() > RDMA_NOTIFY_MAX_NUM) || (notifyAddrKey.size() < RDMA_NOTIFY_MIN_NUM)
466 0 : || (notifyAddrKey.size() > RDMA_NOTIFY_MAX_NUM)
467 0 : || ((signalInfos.size() - RDMA_NOTIFY_MIN_NUM) % linkRoce.qpsPerConnection) != 0
468 0 : || ((notifyAddrKey.size() - RDMA_NOTIFY_MIN_NUM) % linkRoce.qpsPerConnection) != 0) {
469 0 : return HCCL_E_INTERNAL;
470 : }
471 0 : u64 notifyNum = (notifyAddrKey.size() - RDMA_NOTIFY_MIN_NUM) / linkRoce.qpsPerConnection
472 0 : - static_cast<u32>(linkRoce.qpsPerConnection > 1);
473 0 : linkRoce.singleQPNotifyNum = notifyNum;
474 :
475 0 : u64 len = signalInfos.size() * sizeof(HcclSignalInfo);
476 0 : DeviceMem localNotifyListMem;
477 0 : CHK_RET(CopyVectorToDeviceMem(len, localNotifyListMem, signalInfos));
478 0 : linkRoce.localNotifyList = reinterpret_cast<u64>(localNotifyListMem.ptr());
479 0 : channelParamMemList_.emplace_back(std::move(localNotifyListMem));
480 :
481 0 : len = notifyAddrKey.size() * sizeof(AddrKey);
482 0 : DeviceMem remoteNotifyListMem;
483 0 : CHK_RET(CopyVectorToDeviceMem(len, remoteNotifyListMem, notifyAddrKey));
484 0 : linkRoce.remoteNotifyList = reinterpret_cast<u64>(remoteNotifyListMem.ptr());
485 0 : channelParamMemList_.emplace_back(std::move(remoteNotifyListMem));
486 :
487 0 : remoteRes.roceNotifyNum = linkRoce.singleQPNotifyNum;
488 0 : remoteRes.qpNum = linkRoce.qpsPerConnection;
489 :
490 0 : return HCCL_SUCCESS;
491 0 : }
492 :
493 0 : HcclResult ChannelManager::DeepCopyH2DchannelParam(
494 : const HcclIndOpChannelRemoteResV3& hostChannelParam, HcclIndOpChannelRemoteResV3& deviceChannelParam)
495 : {
496 0 : deviceChannelParam = hostChannelParam;
497 : // 拷贝remoteResV2
498 :
499 0 : if (hostChannelParam.remoteResV2 != nullptr && hostChannelParam.listNum > 0) {
500 : // 为设备端的remoteResV2数组分配内存（注意：这个数组存放的是HcclIndOpChannelRemoteResV2结构体）
501 0 : size_t remoteResV2ArraySize = sizeof(HcclIndOpChannelRemoteResV2) * hostChannelParam.listNum;
502 0 : std::shared_ptr<DeviceMem> deviceRemoteResV2Array;
503 0 : EXCEPTION_CATCH(
504 : (deviceRemoteResV2Array = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteResV2ArraySize))),
505 : return HCCL_E_PTR);
506 :
507 : // 为每个数组元素进行深度拷贝，并保存设备内存和主机结构体（指针已调整）
508 0 : std::vector<DeviceMem> elementMemories; // 保存每个元素分配的设备内存（包括内部指针数据）
509 0 : std::vector<HcclIndOpChannelRemoteResV2> hostRemoteResV2Array(hostChannelParam.listNum);
510 :
511 0 : for (uint32_t i = 0; i < hostChannelParam.listNum; ++i) {
512 0 : HcclIndOpChannelRemoteResV2 hostElement = hostChannelParam.remoteResV2[i];
513 0 : HcclIndOpChannelRemoteResV2 deviceElement;
514 : // 深度拷贝一个元素到设备内存，并返回设备内存中的结构体布局（host端）
515 0 : CHK_RET(DeepCopyH2DChannelRemoteResV2(hostElement, deviceElement));
516 : // 保存调整后的主机端结构体（其指针指向设备内存）
517 0 : hostRemoteResV2Array[i] = deviceElement;
518 : }
519 :
520 : // 将主机端的结构体数组（指针已调整）拷贝到设备内存数组
521 0 : CHK_RET(hrtMemSyncCopy(
522 : deviceRemoteResV2Array.get()->ptr(), remoteResV2ArraySize, hostRemoteResV2Array.data(),
523 : remoteResV2ArraySize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
524 :
525 : // 更新设备端参数中的remoteResV2指针
526 : deviceChannelParam.remoteResV2
527 0 : = reinterpret_cast<HcclIndOpChannelRemoteResV2*>(deviceRemoteResV2Array.get()->ptr());
528 0 : channelParamMemVector_.push_back(std::move(deviceRemoteResV2Array));
529 0 : } else {
530 0 : HCCL_ERROR("[%s]invalid hostChannelParam", __func__);
531 0 : return HCCL_E_INTERNAL;
532 : }
533 0 : return HCCL_SUCCESS;
534 : }
535 :
536 0 : HcclResult ChannelManager::DeepCopyH2DChannelRemoteResV2(
537 : const HcclIndOpChannelRemoteResV2& hostRemoteResV2, HcclIndOpChannelRemoteResV2& deviceRemoteResV2)
538 : {
539 : // 复制基本成员
540 0 : deviceRemoteResV2 = hostRemoteResV2;
541 : // 根据通信类型处理不同的通道
542 0 : if (hostRemoteResV2.isUsedRdma) {
543 : // 处理RoCE通道
544 0 : CHK_RET(DeepCopyH2DChannelRoce(hostRemoteResV2.channelRoce, deviceRemoteResV2.channelRoce));
545 : } else {
546 : // 处理P2P通道
547 0 : CHK_RET(DeepCopyH2DChannelP2p(hostRemoteResV2.channelP2p, deviceRemoteResV2.channelP2p));
548 : }
549 0 : return HCCL_SUCCESS;
550 : }
551 :
552 : HcclResult
553 0 : ChannelManager::DeepCopyH2DChannelRoce(const HcclChannelRoce& hostChannelRoce, HcclChannelRoce& deviceChannelRoce)
554 : {
555 : // 复制基本成员
556 0 : deviceChannelRoce = hostChannelRoce;
557 : // 处理remoteUserHostMem
558 0 : if (hostChannelRoce.remoteUserHostMem != nullptr && hostChannelRoce.remoteUserHostMemCount > 0) {
559 0 : size_t remoteUserHostMemSize = hostChannelRoce.remoteUserHostMemCount * sizeof(MemDetails);
560 0 : std::shared_ptr<DeviceMem> deviceMem;
561 0 : EXCEPTION_CATCH(
562 : (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserHostMemSize))), return HCCL_E_PTR);
563 0 : CHK_RET(hrtMemSyncCopy(
564 : deviceMem.get()->ptr(), remoteUserHostMemSize, hostChannelRoce.remoteUserHostMem, remoteUserHostMemSize,
565 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
566 0 : deviceChannelRoce.remoteUserHostMem = reinterpret_cast<MemDetails*>(deviceMem.get()->ptr());
567 0 : channelParamMemVector_.push_back(std::move(deviceMem));
568 0 : } else {
569 0 : deviceChannelRoce.remoteUserHostMem = nullptr;
570 : }
571 : // 处理remoteUserDeviceMem
572 0 : if (hostChannelRoce.remoteUserDeviceMem != nullptr && hostChannelRoce.remoteUserDeviceMemCount > 0) {
573 0 : size_t remoteUserDeviceMemSize = hostChannelRoce.remoteUserDeviceMemCount * sizeof(MemDetails);
574 0 : std::shared_ptr<DeviceMem> deviceMem;
575 0 : EXCEPTION_CATCH(
576 : (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserDeviceMemSize))), return HCCL_E_PTR);
577 0 : CHK_RET(hrtMemSyncCopy(
578 : deviceMem.get()->ptr(), remoteUserDeviceMemSize, hostChannelRoce.remoteUserDeviceMem,
579 : remoteUserDeviceMemSize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
580 0 : deviceChannelRoce.remoteUserDeviceMem = reinterpret_cast<MemDetails*>(deviceMem.get()->ptr());
581 0 : channelParamMemVector_.push_back(std::move(deviceMem));
582 0 : } else {
583 0 : deviceChannelRoce.remoteUserDeviceMem = nullptr;
584 : }
585 :
586 0 : return HCCL_SUCCESS;
587 : }
588 :
589 0 : HcclResult ChannelManager::DeepCopyH2DChannelP2p(const HcclChannelP2p& hostChannelP2p, HcclChannelP2p& deviceChannelP2p)
590 : {
591 : // 复制基本成员
592 0 : deviceChannelP2p = hostChannelP2p;
593 : // 处理remoteUserMem
594 0 : if (hostChannelP2p.remoteUserMem != nullptr && hostChannelP2p.remoteUserMemCount > 0) {
595 0 : size_t remoteUserMemSize = hostChannelP2p.remoteUserMemCount * sizeof(HcclMem);
596 0 : std::shared_ptr<DeviceMem> deviceMem;
597 0 : EXCEPTION_CATCH(
598 : (deviceMem = std::make_shared<DeviceMem>(DeviceMem::alloc(remoteUserMemSize))), return HCCL_E_PTR);
599 0 : CHK_RET(hrtMemSyncCopy(
600 : deviceMem.get()->ptr(), remoteUserMemSize, hostChannelP2p.remoteUserMem, remoteUserMemSize,
601 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
602 0 : deviceChannelP2p.remoteUserMem = reinterpret_cast<HcclMem*>(deviceMem.get()->ptr());
603 0 : channelParamMemVector_.push_back(std::move(deviceMem));
604 0 : } else {
605 0 : deviceChannelP2p.remoteUserMem = nullptr;
606 : }
607 0 : return HCCL_SUCCESS;
608 : }
609 :
610 0 : HcclResult ChannelManager::ReleaseChannelParam(HcclIndOpChannelRemoteResV3& channelParam)
611 : {
612 : // 释放remoteResV2
613 0 : if (channelParam.remoteResV2 != nullptr) {
614 0 : for (uint32_t i = 0; i < channelParam.listNum; ++i) {
615 0 : HcclIndOpChannelRemoteResV2& remoteRes = channelParam.remoteResV2[i];
616 0 : if (remoteRes.isUsedRdma) {
617 0 : if (remoteRes.channelRoce.remoteUserHostMem != nullptr) {
618 0 : free(remoteRes.channelRoce.remoteUserHostMem);
619 : }
620 0 : if (remoteRes.channelRoce.remoteUserDeviceMem != nullptr) {
621 0 : free(remoteRes.channelRoce.remoteUserDeviceMem);
622 : }
623 : } else {
624 0 : if (remoteRes.channelP2p.remoteUserMem != nullptr) {
625 0 : free(remoteRes.channelP2p.remoteUserMem);
626 : }
627 : }
628 : }
629 : }
630 0 : free(channelParam.remoteResV2);
631 0 : channelParam.remoteResV2 = nullptr;
632 :
633 : // 将kernel下发时临时分配的deviceMem一起销毁
634 0 : channelParamMemVector_.clear();
635 0 : channelParamMemList_.clear();
636 0 : return HCCL_SUCCESS;
637 : }
638 :
639 0 : HcclResult ChannelManager::AicpuChannelInit(
640 : const std::string& commId, const std::string& tag, CommEngine engine, const OpCommTransport& opTransportResponse,
641 : ChannelHandle* channelList, uint32_t listNum)
642 : {
643 0 : HcclIndOpChannelRemoteResV3 channelParam{};
644 0 : CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam)));
645 0 : uint64_t beginTime = hrtMsprofSysCycleTime();
646 : // channelParam资源参数填充
647 0 : strncpy_s(channelParam.hcomId, HCOMID_MAX_LENGTH, commId.c_str(), HCOMID_MAX_LENGTH - 1);
648 0 : strncpy_s(channelParam.channelTag, TAG_MAX_LENGTH, tag.c_str(), TAG_MAX_LENGTH - 1);
649 0 : channelParam.engine = engine;
650 0 : channelParam.localUserRank = userRank_;
651 0 : channelParam.multiQpThreshold = GetExternalInputMultiQpThreshold();
652 :
653 : // 为device侧的channelList分配内存
654 0 : DeviceMem deviceChannelList = DeviceMem::alloc(listNum * sizeof(ChannelHandle));
655 0 : CHK_PTR_NULL(deviceChannelList.ptr());
656 0 : channelParam.channelList = static_cast<void*>(deviceChannelList.ptr());
657 0 : channelParam.listNum = listNum;
658 :
659 : // 将建链获取的远端数据填充到channelParam
660 0 : HcclResult ret = ParseChannelRemoteDataToMem(opTransportResponse, channelParam);
661 0 : if (ret != HCCL_SUCCESS) {
662 0 : HCCL_ERROR("[%s] ParseChannelRemoteDataToMem failed, return [%d].", __func__, ret);
663 0 : ReleaseChannelParam(channelParam);
664 0 : return ret;
665 : }
666 :
667 : // 创建局部流
668 0 : Stream localStream(StreamType::STREAM_TYPE_ONLINE);
669 0 : constexpr u32 aicpuStreamMode = 1;
670 0 : CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode));
671 :
672 : // 将channelParam内部的host内存拷贝成device内存
673 0 : HcclIndOpChannelRemoteResV3 deviceChannelParam = channelParam;
674 0 : CHK_RET(DeepCopyH2DchannelParam(channelParam, deviceChannelParam));
675 :
676 0 : DeviceMem addr = DeviceMem::alloc(sizeof(deviceChannelParam));
677 0 : CHK_PTR_NULL(addr.ptr());
678 0 : CHK_RET(hrtMemSyncCopy(
679 : addr.ptr(), sizeof(deviceChannelParam), &deviceChannelParam, sizeof(deviceChannelParam),
680 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
681 :
682 : // 下kernel
683 0 : std::string kernelName = "RunAicpuIndOpChannelInit";
684 : struct InitTask {
685 : u64 context;
686 : bool isCustom;
687 : };
688 0 : InitTask customInitTask = {0};
689 0 : customInitTask.context = reinterpret_cast<u64>(addr.ptr());
690 0 : customInitTask.isCustom = false;
691 :
692 0 : u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
693 : std::numeric_limits<uint16_t>::max() :
694 : NOTIFY_DEFAULT_WAIT_TIME;
695 0 : CHK_RET(AicpuAclKernelLaunch(
696 : localStream.ptr(), reinterpret_cast<void*>(&customInitTask), sizeof(customInitTask), binHandle_, kernelName,
697 : true, timeOut));
698 0 : CHK_RET(hcclStreamSynchronize(localStream.ptr(), CommConfiger::GetInstance().GetCommConfigExecTimeOut(tag)));
699 :
700 : // 将device侧的channelList拷贝回host侧的channelList
701 0 : CHK_RET(hrtMemSyncCopy(
702 : channelList, listNum * sizeof(ChannelHandle), deviceChannelList.ptr(), listNum * sizeof(ChannelHandle),
703 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
704 :
705 : // 手动释放channelParam中申请的内存
706 0 : CHK_RET(ReleaseChannelParam(channelParam));
707 0 : const std::string profName = "RunAicpuIndOpChannelInit";
708 0 : HCCL_DEBUG("[%s] RunAicpuIndOpChannelInit", __func__);
709 : // 上报初始化kernel的时间
710 0 : HcommProfilingReportKernel(beginTime, profName.c_str());
711 0 : return HCCL_SUCCESS;
712 0 : }
713 :
714 : const std::map<CommEngine, std::string> COMM_ENGINE_TYPE_STR_MAP{
715 : {CommEngine::COMM_ENGINE_CPU, "host_cpu"}, {CommEngine::COMM_ENGINE_CPU_TS, "host_cpu_ts"},
716 : {CommEngine::COMM_ENGINE_AICPU, "aicpu"}, {CommEngine::COMM_ENGINE_AICPU_TS, "aicpu_ts"},
717 : {CommEngine::COMM_ENGINE_AIV, "aiv"}, {CommEngine::COMM_ENGINE_CCU, "ccu"},
718 : {CommEngine::COMM_ENGINE_RESERVED, "reserved"}};
719 :
720 0 : std::string GetCommEngineEnumStr(CommEngine engine)
721 : {
722 0 : auto iter = COMM_ENGINE_TYPE_STR_MAP.find(engine);
723 0 : if (iter == COMM_ENGINE_TYPE_STR_MAP.end()) {
724 0 : return "CommEngine=" + std::to_string(engine);
725 : } else {
726 0 : return iter->second;
727 : }
728 : }
729 :
730 1 : HcclResult ChannelManager::ChannelCommCreate(
731 : const std::string& commId, CommEngine engine, const HcclChannelDesc* channelDescList, uint32_t listNum,
732 : ChannelHandle* channelList)
733 : {
734 1 : CHK_RET(CheckChannelParam(engine, channelDescList, listNum));
735 :
736 : // channel复用，以tag + engine + remoterank + channelProtocol 作为channel标识
737 0 : std::vector<HcclChannelDesc> needCreateDescs;
738 0 : std::vector<uint32_t> needCreateIndices;
739 0 : std::string tag = commId;
740 0 : CHK_RET(PrepareHandleArray(tag, engine, channelDescList, listNum, channelList, needCreateDescs, needCreateIndices));
741 :
742 : // 对未复用的channelDesc进行建链
743 0 : if (needCreateDescs.size() > 0) {
744 : // 构造建链param
745 0 : OpCommTransport opCommTransport = BuildChannelRequests(needCreateDescs);
746 0 : std::string linkTag = commId + "_" + GetCommEngineEnumStr(engine);
747 0 : bool isAicpuModeEn = false;
748 0 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
749 0 : isAicpuModeEn = true;
750 : }
751 0 : std::vector<HcclMemHandle> memHandles = CollectMemHandles(needCreateDescs);
752 0 : CHK_RET(channelCallbacks_.indOpTransportAlloc(
753 : linkTag, opCommTransport, isAicpuModeEn, memHandles.data(), static_cast<uint32_t>(memHandles.size())));
754 :
755 0 : uint32_t level0 = 0;
756 0 : std::vector<LINK> links = opCommTransport[level0][level0].links;
757 0 : uint32_t newDescNum = needCreateDescs.size();
758 : // 创建host或device侧channel句柄
759 0 : if (isAicpuModeEn) {
760 : // Kernel下发恢复
761 0 : if (!callbacks_.getAicpuCommState()) {
762 0 : HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
763 0 : CHK_PRT_RET(
764 : ret != HCCL_SUCCESS,
765 : HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
766 0 : callbacks_.setAicpuCommState(true);
767 : }
768 0 : std::unique_ptr<ChannelHandle[]> tmpChannelList = std::make_unique<ChannelHandle[]>(newDescNum);
769 0 : CHK_RET(AicpuChannelInit(commId, tag, engine, opCommTransport, tmpChannelList.get(), newDescNum));
770 0 : for (u32 i = 0; i < newDescNum; i++) {
771 0 : uint32_t arrayIndex = needCreateIndices[i];
772 0 : channelList[arrayIndex] = tmpChannelList[i];
773 0 : CHK_RET(RegisterHandle(tag, engine, needCreateDescs[i], tmpChannelList[i]));
774 0 : ChannelHandle channelHandle = reinterpret_cast<ChannelHandle>(links[i].get());
775 0 : CHK_RET(RegisterHandleHDPair(tmpChannelList[i], channelHandle));
776 : }
777 0 : } else {
778 0 : for (u32 i = 0; i < newDescNum; i++) {
779 0 : uint32_t arrayIndex = needCreateIndices[i];
780 0 : ChannelHandle channelHandle = reinterpret_cast<ChannelHandle>(links[i].get());
781 0 : channelList[arrayIndex] = channelHandle;
782 0 : CHK_RET(RegisterHandle(tag, engine, needCreateDescs[i], channelHandle));
783 : }
784 : }
785 : // 保存link
786 0 : for (auto& link : links) {
787 : // 设置成员变量保存link
788 0 : channelLinks_.push_back(link);
789 : }
790 0 : }
791 0 : return HCCL_SUCCESS;
792 0 : }
793 :
794 0 : HcclResult ChannelManager::ChannelCommGetNotifyNum(ChannelHandle channel, uint32_t* notifyNum)
795 : {
796 0 : CHK_RET(IsChannelExist(channel));
797 : ChannelHandle hostchannel;
798 0 : CHK_RET(GetHostChannel(channel, hostchannel));
799 :
800 0 : Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
801 0 : *notifyNum = transportPtr->GetNotifyNum();
802 0 : return HCCL_SUCCESS;
803 : }
804 :
805 3 : HcclResult ChannelManager::ChannelCommDestroy(ChannelHandle* channelList, uint32_t channelNum)
806 : {
807 7 : for (uint32_t i = 0; i < channelNum; ++i) {
808 4 : UnregisterHandle(channelList[i]);
809 4 : channelList[i] = 0;
810 : }
811 3 : return HCCL_SUCCESS;
812 : }
813 :
814 0 : HcclResult ChannelManager::ChannelCommGetHcclBuffer(ChannelHandle channel, CommBuffer* buffer)
815 : {
816 : ChannelHandle hostchannel;
817 0 : CHK_RET(IsChannelExist(channel));
818 0 : CHK_RET(GetHostChannel(channel, hostchannel));
819 0 : Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
820 :
821 0 : buffer->addr = nullptr;
822 0 : CHK_RET(transportPtr->GetRemoteMem(UserMemType::INPUT_MEM, &buffer->addr));
823 0 : CHK_PTR_NULL(buffer->addr);
824 0 : u64 tempSize = 0;
825 0 : CHK_RET(transportPtr->GetRemoteMemSize(UserMemType::INPUT_MEM, tempSize));
826 0 : buffer->size = static_cast<uint64_t>(tempSize);
827 0 : buffer->type = HCCL_MEM_TYPE_DEVICE;
828 0 : HCCL_INFO(
829 : "[%s]channel[%llu] channelKey[%s] get remote hccl buffer success, remote addr[%p], size[%llu]", __func__,
830 : channel, keyMap_[channel].c_str(), buffer->addr, buffer->size);
831 0 : return HCCL_SUCCESS;
832 : }
833 :
834 0 : HcclResult ChannelManager::ChannelCommGetRemoteMem(ChannelHandle channel, HcclMem** remoteMem, uint32_t* memNum)
835 : {
836 0 : CHK_RET(IsChannelExist(channel));
837 : ChannelHandle hostchannel;
838 0 : CHK_RET(GetHostChannel(channel, hostchannel));
839 0 : Transport* transportPtr = reinterpret_cast<Transport*>(hostchannel);
840 :
841 0 : CHK_RET(transportPtr->GetIndOpRemoteMem(remoteMem, memNum));
842 0 : HCCL_INFO("[%s]get remote mem success, mem num[%u]", __func__, *memNum);
843 0 : return HCCL_SUCCESS;
844 : }
845 :
846 404 : HcclResult ChannelManager::ReleaseChannel()
847 : {
848 404 : for (auto& link : channelLinks_) {
849 0 : if (link != nullptr) {
850 0 : if (link->DeInit() != HCCL_SUCCESS) {
851 0 : HCCL_ERROR("[%s]transport[%p] deinit failed.", __func__, link.get());
852 : }
853 : }
854 : }
855 405 : channelLinks_.clear();
856 405 : return HCCL_SUCCESS;
857 : }
858 :
859 236 : HcclResult ChannelManager::SetHcclQos(u32 hcclQos)
860 : {
861 236 : HCCL_INFO("[ChannelManager] [SetHcclQos] hcclQos[%u]", hcclQos);
862 236 : hcclQos_ = hcclQos;
863 236 : return HCCL_SUCCESS;
864 : }
865 : } // namespace hccl
|
