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