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