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_process.h"
12 : #include <cstdint>
13 : #include <map>
14 : #include <memory>
15 : #include <vector>
16 : #include "exception_handler.h"
17 : #include "channel_param.h"
18 : #include "channel.h"
19 : #include "aicpu_ts_channel_helper.h"
20 : #include "aiv_channel_helper.h"
21 : #include "aicpu_ts_urma_channel.h"
22 : #include "aicpu_ts_uboe_channel.h"
23 : #include "aicpu_ts_ubg_channel.h"
24 : #include "aicpu_ts_roce_channel_v2.h"
25 : #include "acl/acl_rt.h"
26 : #include "launch_aicpu.h"
27 : #include "hcclCommDfx.h"
28 : #include "env_config/env_config.h"
29 : #include "aicpu_ts_p2p_channel.h"
30 : #include "aiv_urma_channel.h"
31 : #include "mem_device_pub.h"
32 : #include "comm_engine_utils.h"
33 :
34 : namespace hcomm {
35 :
36 : std::unordered_map<ChannelHandle, std::shared_ptr<Channel>> ChannelProcess::g_ChannelMap;
37 : std::unordered_map<DeviceChannelKey, ChannelHandle, DeviceChannelKeyHash> ChannelProcess::g_ChannelD2HMap;
38 : std::mutex ChannelProcess::g_ChannelMapMtx;
39 :
40 : template <typename Func>
41 42 : HcclResult ChannelProcess::WithChannelByHandleLocked(ChannelHandle inHandle, Func &&func)
42 : {
43 42 : int32_t deviceId = 0;
44 42 : CHK_RET(hrtGetDevice(&deviceId));
45 :
46 42 : std::shared_ptr<Channel> channelPtr = nullptr;
47 : {
48 : // 单锁：该锁同时保护 g_ChannelD2HMap 和 g_ChannelMap
49 42 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
50 :
51 : // 1) D2H 映射
52 42 : DeviceChannelKey key{deviceId, inHandle};
53 42 : auto itH = g_ChannelD2HMap.find(key);
54 42 : if (itH == g_ChannelD2HMap.end()) {
55 4 : HCCL_ERROR("[%s] handle not found in g_ChannelD2HMap, deviceId[%d], inHandle[0x%llx].", __func__, deviceId, inHandle);
56 4 : return HcclResult::HCCL_E_NOT_FOUND;
57 : }
58 38 : const ChannelHandle mappedHandle = itH->second;
59 :
60 : // 2) ChannelMap 查找
61 38 : auto itC = g_ChannelMap.find(mappedHandle);
62 38 : if (itC == g_ChannelMap.end() || !itC->second) {
63 0 : HCCL_ERROR("[%s] channel not found in g_ChannelMap, deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx].",
64 : __func__,
65 : deviceId,
66 : inHandle,
67 : mappedHandle);
68 0 : return HcclResult::HCCL_E_INTERNAL;
69 : }
70 :
71 38 : channelPtr = itC->second;
72 38 : if (channelPtr == nullptr) {
73 0 : HCCL_ERROR(
74 : "[%s] null channel pointer, deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx].", __func__, deviceId, inHandle, mappedHandle);
75 0 : return HcclResult::HCCL_E_INTERNAL;
76 : }
77 42 : }
78 :
79 : // 3) 执行用户逻辑
80 38 : return std::forward<Func>(func)(*channelPtr);
81 42 : }
82 :
83 18 : HcclResult ChannelProcess::CreateChannelsLoop(EndpointHandle endpointHandle, CommEngine engine,
84 : HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *outHandles)
85 : {
86 18 : CHK_PTR_NULL(endpointHandle);
87 :
88 17 : int32_t deviceId = 0;
89 17 : CHK_RET(hrtGetDevice(&deviceId));
90 :
91 34 : for (uint32_t i = 0; i < channelNum; ++i) {
92 17 : std::shared_ptr<Channel> tmpPtr = nullptr;
93 17 : CHK_RET_UNAVAIL(Channel::CreateChannel(endpointHandle, engine, channelDescs[i], tmpPtr));
94 17 : CHK_SMART_PTR_NULL(tmpPtr);
95 :
96 17 : ChannelHandle handle = reinterpret_cast<ChannelHandle>(tmpPtr.get());
97 17 : outHandles[i] = handle;
98 17 : HCCL_INFO("%s deviceId[%d], handle[0x%llx], ptr[%p]", __func__, deviceId, handle, tmpPtr.get());
99 :
100 : // 仅在修改全局表时持锁
101 : {
102 17 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
103 :
104 17 : if (g_ChannelMap.find(handle) != g_ChannelMap.end()) {
105 0 : HCCL_ERROR("[%s] channel handle already exists [0x%llx] in ChannelMap", __func__, handle);
106 0 : return HCCL_E_INTERNAL;
107 : }
108 17 : DeviceChannelKey key{deviceId, handle};
109 17 : if (g_ChannelD2HMap.find(key) != g_ChannelD2HMap.end()) {
110 0 : HCCL_ERROR("[%s] channel handle already exists deviceId[%d], handle[0x%llx] in g_ChannelD2HMap", __func__, deviceId, handle);
111 0 : return HCCL_E_INTERNAL;
112 : }
113 :
114 17 : g_ChannelMap.emplace(handle, std::move(tmpPtr));
115 17 : g_ChannelD2HMap.emplace(key, handle);
116 17 : }
117 17 : }
118 17 : return HCCL_SUCCESS;
119 : }
120 :
121 0 : HcclResult ChannelProcess::ChannelUpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum, ChannelHandle channelHandle)
122 : {
123 : EXCEPTION_HANDLE_BEGIN
124 0 : int32_t deviceId = 0;
125 0 : CHK_RET(hrtGetDevice(&deviceId));
126 :
127 0 : Channel *channel = nullptr;
128 : {
129 0 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
130 : // 1) D2H 映射
131 0 : DeviceChannelKey key{deviceId, channelHandle};
132 0 : auto itH = g_ChannelD2HMap.find(key);
133 0 : if (itH == g_ChannelD2HMap.end()) {
134 0 : HCCL_ERROR("[%s] handle not found in g_ChannelD2HMap, deviceId[%d], channelHandle[0x%llx].", __func__, deviceId, channelHandle);
135 0 : return HcclResult::HCCL_E_NOT_FOUND;
136 : }
137 0 : const ChannelHandle mappedHandle = itH->second;
138 :
139 : // 2) ChannelMap 查找
140 0 : auto itC = g_ChannelMap.find(mappedHandle);
141 0 : if (itC == g_ChannelMap.end() || !itC->second) {
142 0 : HCCL_ERROR("[%s] channel not found in g_ChannelMap, deviceId[%d], channelHandle[0x%llx], mappedHandle[0x%llx].",
143 : __func__,
144 : deviceId,
145 : channelHandle,
146 : mappedHandle);
147 0 : return HcclResult::HCCL_E_INTERNAL;
148 : }
149 0 : channel = itC->second.get();
150 0 : }
151 : // UpdateMemInfo需要rank间交互，若在锁内执行会导致单进程多线程场景其他rank被锁拦住
152 0 : CHK_RET(channel->UpdateMemInfo(memHandles, memHandleNum));
153 0 : EXCEPTION_HANDLE_END
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 10 : HcclResult ChannelProcess::ChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t *statusList)
158 : {
159 : EXCEPTION_HANDLE_BEGIN
160 :
161 : // 不得随意添加无效日志，可能造成刷屏
162 10 : CHK_PTR_NULL(channelList);
163 9 : CHK_PTR_NULL(statusList);
164 :
165 8 : u32 readyCount = 0;
166 8 : u32 failCount = 0;
167 :
168 18 : for (uint32_t i = 0; i < listNum; ++i) {
169 11 : const ChannelHandle inHandle = channelList[i];
170 11 : int32_t status = 0;
171 : // 当前通道状态如果已为FAILED/SOCKET_TIMEOUT，说明前面已经失败过，无需再重新获取状态，继续轮询下一个通道，避免日志刷屏
172 11 : if (statusList[i] == ChannelStatus::FAILED || statusList[i] == ChannelStatus::SOCKET_TIMEOUT) {
173 1 : failCount++;
174 1 : continue;
175 : }
176 : // 单锁：D2H 映射 + 查 map + 锁内调用 GetStatus()
177 10 : HcclResult ret = WithChannelByHandleLocked(inHandle, [&status](Channel &channel) -> HcclResult {
178 9 : status = channel.GetStatus(); // 锁内调用，防止 destroy 并发释放
179 9 : return HcclResult::HCCL_SUCCESS;
180 : });
181 10 : if (ret != HcclResult::HCCL_SUCCESS) {
182 1 : HCCL_ERROR("[%s] Get ChannelHandle failed.", __func__);
183 1 : return ret;
184 : }
185 : // 某一个channel状态为FAILED/SOCKET_TIMEOUT时不直接返回，否则后面的channel无法轮询完，状态无法到达终态；
186 9 : if (status == ChannelStatus::FAILED) {
187 2 : HCCL_ERROR("[%s] FAILED, channel idx[%u], status[%d]", __func__, i, status);
188 2 : failCount++;
189 : }
190 9 : if (status == ChannelStatus::SOCKET_TIMEOUT) {
191 1 : HCCL_ERROR("[%s] TIMEOUT, channel idx[%u], status[%d]", __func__, i, status);
192 1 : failCount++;
193 : }
194 :
195 9 : readyCount += (status == ChannelStatus::READY) ? 1 : 0;
196 9 : statusList[i] = status;
197 : }
198 7 : if (readyCount + failCount < listNum) {
199 2 : return HCCL_E_AGAIN;
200 : }
201 5 : if (readyCount != listNum) {
202 4 : HCCL_ERROR("[%s] NETWORK, readyCount[%u], failCount[%u], listNum[%u]", __func__, readyCount, failCount, listNum);
203 4 : return HCCL_E_NETWORK;
204 : }
205 0 : EXCEPTION_HANDLE_END
206 1 : return HCCL_SUCCESS;
207 : }
208 :
209 31 : HcclResult ChannelProcess::GetChannelsInfo(const ChannelHandle *channelList, uint32_t listNum,
210 : std::vector<CommEngine> &engines, std::vector<HcommChannelDesc> &channelDescs,
211 : std::vector<ChannelStatus> &statusList)
212 : {
213 31 : CHK_PTR_NULL(channelList);
214 30 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", __func__, listNum), HCCL_E_PARA);
215 :
216 29 : channelDescs.resize(listNum);
217 29 : statusList.resize(listNum);
218 29 : engines.resize(listNum);
219 58 : for (uint32_t i = 0; i < listNum; ++i) {
220 64 : HcclResult ret = WithChannelByHandleLocked(channelList[i],
221 32 : [i, &engines, &channelDescs, &statusList](Channel &channel) -> HcclResult {
222 29 : engines[i] = channel.GetEngine();
223 29 : channelDescs[i] = channel.GetChannelDesc();
224 29 : statusList[i] = channel.GetStatus();
225 29 : return HcclResult::HCCL_SUCCESS;
226 : });
227 32 : if (ret != HcclResult::HCCL_SUCCESS) {
228 3 : HCCL_ERROR("[%s] Get channel[%u] info failed.", __func__, i);
229 3 : return ret;
230 : }
231 29 : if (statusList[i] == ChannelStatus::FAILED || statusList[i] == ChannelStatus::SOCKET_TIMEOUT) {
232 5 : HCCL_RUN_WARNING("[%s] FAILED, channel idx[%u], status[%d]", __func__, i, statusList[i]);
233 : }
234 : }
235 26 : HCCL_DEBUG("[%s] SUCCESS.", __func__);
236 26 : return HCCL_SUCCESS;
237 : }
238 :
239 25 : void ConvertToLinkStatus(const std::vector<ChannelStatus> &internalStatus,
240 : std::vector<int32_t> &linkStatusList)
241 : {
242 53 : for (size_t i = 0; i < internalStatus.size(); i++) {
243 28 : switch (internalStatus[i]) {
244 4 : case ChannelStatus::FAILED:
245 4 : linkStatusList[i] = HCOMM_CHANNEL_STATUS_FAILED;
246 4 : break;
247 1 : case ChannelStatus::SOCKET_TIMEOUT:
248 1 : linkStatusList[i] = HCOMM_CHANNEL_STATUS_TIMEOUT;
249 1 : break;
250 4 : case ChannelStatus::READY:
251 4 : linkStatusList[i] = HCOMM_CHANNEL_STATUS_READY;
252 4 : break;
253 19 : default:
254 19 : linkStatusList[i] = HCOMM_CHANNEL_STATUS_CONNECTING;
255 19 : break;
256 : }
257 : }
258 25 : }
259 :
260 17 : void CopyLinkStatusToOutput(const std::vector<int32_t> &linkStatusList,
261 : int32_t *statusList, uint32_t listNum)
262 : {
263 34 : for (uint32_t i = 0; i < listNum; i++) {
264 17 : statusList[i] = linkStatusList[i];
265 : }
266 17 : }
267 :
268 25 : HcclResult ChannelProcess::HandleStatusByEngine(const ChannelHandle *channelList, uint32_t listNum,
269 : const std::vector<CommEngine> &engines, const std::vector<HcommChannelDesc> &channelDescFinals,
270 : const std::vector<ChannelStatus> &internalStatus, int32_t *statusList)
271 : {
272 25 : std::vector<int32_t> linkStatusList(listNum);
273 25 : ConvertToLinkStatus(internalStatus, linkStatusList);
274 :
275 25 : std::map<CommEngine, std::vector<uint32_t>> groups;
276 53 : for (uint32_t i = 0; i < listNum; i++) {
277 28 : groups[engines[i]].push_back(i);
278 : }
279 :
280 52 : for (auto &entry : groups) {
281 27 : CommEngine engine = entry.first;
282 27 : const std::vector<uint32_t> &indices = entry.second;
283 27 : uint32_t subNum = static_cast<uint32_t>(indices.size());
284 54 : std::vector<ChannelHandle> subChannels(subNum);
285 54 : std::vector<HcommChannelDesc> subDescs(subNum);
286 27 : std::vector<int32_t> subLinkStatus(subNum);
287 55 : for (uint32_t j = 0; j < subNum; j++) {
288 28 : subChannels[j] = channelList[indices[j]];
289 28 : subDescs[j] = channelDescFinals[indices[j]];
290 28 : subLinkStatus[j] = linkStatusList[indices[j]];
291 : }
292 :
293 27 : std::vector<int32_t> subStatus(subNum);
294 27 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
295 6 : CHK_RET(AicpuTsChannelHelper::HandleStatus(
296 : subChannels.data(), subNum, engine, subDescs.data(), subLinkStatus, subStatus.data()));
297 27 : } else if (engine == COMM_ENGINE_AIV) {
298 4 : CHK_RET(AivChannelHelper::HandleStatus(
299 : subChannels.data(), subNum, subDescs.data(), subLinkStatus, subStatus.data()));
300 : } else {
301 17 : CopyLinkStatusToOutput(subLinkStatus, subStatus.data(), subNum);
302 : }
303 :
304 55 : for (uint32_t j = 0; j < subNum; j++) {
305 28 : statusList[indices[j]] = subStatus[j];
306 : }
307 27 : }
308 :
309 25 : return HCCL_SUCCESS;
310 25 : }
311 :
312 0 : HcclResult ChannelProcess::CombineHostMemory(const std::vector<std::vector<char>> &hostPackBuffers,
313 : hccl::HostMem &hostPackBuf)
314 : {
315 0 : if (hostPackBuffers.empty()) {
316 0 : HCCL_ERROR("[%s] hostPackBuffers is empty, please check.", __func__);
317 0 : return HCCL_E_PARA;
318 : }
319 :
320 : // 将离散数据复制到连续内存中
321 0 : u8 *dstPtr = static_cast<u8 *>(hostPackBuf.ptr()); // 目标内存起始地址
322 0 : u64 dstMax = hostPackBuf.size();
323 0 : u64 packSize = 0;
324 :
325 0 : for (const auto &mem : hostPackBuffers) {
326 0 : packSize += mem.size();
327 0 : CHK_PRT_RET(packSize > dstMax,
328 : HCCL_ERROR("[%s] fail, packSize[%llu] is bigger than dstMax[%llu]", __func__, packSize, dstMax),
329 : HCCL_E_PARA);
330 :
331 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dstPtr, mem.size(), mem.data(), mem.size()));
332 0 : dstPtr += mem.size(); // 移动目标指针
333 : }
334 :
335 0 : HCCL_INFO("[%s] end of merging host memory, hostPackBuf.addr[%p], hostPackBuf.size[%zu]",
336 : __func__,
337 : hostPackBuf.ptr(),
338 : hostPackBuf.size());
339 :
340 0 : return HCCL_SUCCESS;
341 : }
342 :
343 5 : HcclResult ChannelProcess::FillChannelD2HMap(ChannelHandle *deviceChannelHandles,
344 : ChannelHandle *hostChannelHandles, uint32_t listNum)
345 : {
346 5 : CHK_PTR_NULL(deviceChannelHandles);
347 4 : CHK_PTR_NULL(hostChannelHandles);
348 3 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", __func__, listNum), HCCL_E_PARA);
349 :
350 2 : int32_t deviceId = 0;
351 2 : CHK_RET(hrtGetDevice(&deviceId));
352 :
353 2 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
354 4 : for (uint32_t idx = 0; idx < listNum; idx++) {
355 2 : auto deviceChannelHandle = deviceChannelHandles[idx];
356 2 : auto hostChannelHandle = hostChannelHandles[idx];
357 2 : HCCL_INFO("%s deviceId[%d], deviceChannelHandle[0x%llx], hostChannelHandle[0x%llx]",
358 : __func__,
359 : deviceId,
360 : deviceChannelHandle,
361 : hostChannelHandle);
362 2 : DeviceChannelKey key{deviceId, deviceChannelHandle};
363 2 : g_ChannelD2HMap[key] = hostChannelHandle;
364 : }
365 :
366 2 : return HCCL_SUCCESS;
367 2 : }
368 :
369 0 : HcclResult ChannelProcess::RegisterChannelD2HMap(ChannelHandle *deviceChannelHandles,
370 : ChannelHandle *hostChannelHandles, uint32_t listNum)
371 : {
372 0 : return FillChannelD2HMap(deviceChannelHandles, hostChannelHandles, listNum);
373 : }
374 :
375 0 : static HcclResult FillChannelParam(HcclChannelUrmaRes &channelParam,
376 : const std::string &commTag,
377 : hccl::DeviceMem &deviceChannelList,
378 : hccl::DeviceMem &devicePackBuf,
379 : uint32_t listNum,
380 : uint32_t totalListNum,
381 : hccl::DeviceMem &channelSizeAddr)
382 : {
383 : // channelParam资源参数填充
384 0 : s32 sRet = strncpy_s(channelParam.hcomId, HCOMID_MAX_LENGTH, commTag.c_str(), HCOMID_MAX_LENGTH - 1);
385 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] str copy fail. return[%d]", __func__, sRet), HCCL_E_INTERNAL);
386 :
387 0 : channelParam.channelList = static_cast<void *>(deviceChannelList.ptr());
388 0 : channelParam.listNum = listNum;
389 0 : channelParam.uniqueIdAddr = static_cast<void *>(devicePackBuf.ptr());
390 0 : channelParam.uniqueIdSize = totalListNum;
391 0 : channelParam.channelSizeAddr = static_cast<void *>(channelSizeAddr.ptr());
392 :
393 0 : CHK_RET(hrtGetDevice(&channelParam.deviceLogicId));
394 : DevType devType;
395 0 : CHK_RET(hrtGetDeviceType(devType));
396 0 : channelParam.deviceType = static_cast<u32>(devType);
397 :
398 0 : return HCCL_SUCCESS;
399 : }
400 :
401 : template<typename T>
402 0 : static HcclResult LaunchKernelDeviceParam(const T &channelParam, aclrtBinHandle binHandle, const std::string &kernelName)
403 : {
404 0 : hccl::Stream localStream = hccl::Stream(hccl::StreamType::STREAM_TYPE_ONLINE);
405 0 : constexpr u32 aicpuStreamMode = 1;
406 0 : CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode));
407 :
408 0 : hccl::DeviceMem addr = hccl::DeviceMem::alloc(sizeof(T));
409 0 : CHK_PTR_NULL(addr.ptr());
410 :
411 0 : CHK_RET(hrtMemSyncCopy(addr.ptr(),
412 : sizeof(T),
413 : &channelParam,
414 : sizeof(T),
415 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
416 :
417 0 : uint64_t context = reinterpret_cast<uint64_t>(addr.ptr());
418 :
419 0 : CHK_RET(hccl::AicpuAclKernelLaunch(localStream.ptr(),
420 : reinterpret_cast<void *>(&context),
421 : sizeof(context),
422 : binHandle,
423 : kernelName,
424 : true,
425 : NOTIFY_DEFAULT_WAIT_TIME));
426 :
427 0 : constexpr u32 STREAM_SYNC_TIMEOUT_MS = 60;
428 0 : CHK_RET(hcclStreamSynchronize(localStream.ptr(), STREAM_SYNC_TIMEOUT_MS));
429 :
430 0 : HCCL_INFO("[%s] kernel[%s] launch success.", __func__, kernelName.c_str());
431 0 : return HCCL_SUCCESS;
432 0 : }
433 :
434 0 : static HcclResult LaunchKernel(const HcclChannelUrmaRes &channelParam, aclrtBinHandle binHandle, const std::string &kernelName)
435 : {
436 0 : return LaunchKernelDeviceParam(channelParam, binHandle, kernelName);
437 : }
438 :
439 3 : HcclResult ChannelProcess::LaunchChannelKernelCommon(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,
440 : HcommChannelDesc* hcommDesc, uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle,
441 : const std::string &kernelName, bool needProfiling)
442 : {
443 3 : CHK_PTR_NULL(channelHandles);
444 2 : CHK_PTR_NULL(hostChannelHandles);
445 1 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", __func__, listNum), HCCL_E_PARA);
446 :
447 0 : HCCL_RUN_INFO("[%s] listNum[%u], commTag[%s]", __func__, listNum, commTag.c_str());
448 0 : std::vector<std::vector<char>> hostPackBuffers(listNum);
449 0 : HcclChannelUrmaRes channelParam{};
450 0 : CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam)));
451 :
452 : // 获取host侧序列化的地址
453 0 : std::vector<u32> channelSizeVec{};
454 0 : uint32_t totalListNum = 0;
455 0 : for (uint32_t index = 0; index < listNum; index++) {
456 0 : if (hcommDesc[index].remoteEndpoint.protocol == CommProtocol::COMM_PROTOCOL_PCIE) {
457 0 : auto aicpuTsP2pChannel = reinterpret_cast<AicpuTsP2pChannel *>(hostChannelHandles[index]);
458 0 : CHK_PRT(aicpuTsP2pChannel->H2DResPack(hostPackBuffers[index]));
459 0 : } else if (hcommDesc[index].remoteEndpoint.protocol == CommProtocol::COMM_PROTOCOL_UBOE) {
460 0 : auto aicpuTsUboeChannel = reinterpret_cast<AicpuTsUboeChannel *>(hostChannelHandles[index]);
461 0 : CHK_PRT(aicpuTsUboeChannel->H2DResPack(hostPackBuffers[index]));
462 0 : } else if (hcommDesc[index].remoteEndpoint.protocol == CommProtocol::COMM_PROTOCOL_UBG) {
463 0 : auto aicpuTsUbgChannel = reinterpret_cast<AicpuTsUbgChannel *>(hostChannelHandles[index]);
464 0 : CHK_PRT(aicpuTsUbgChannel->H2DResPack(hostPackBuffers[index]));
465 0 : } else if (hcommDesc[index].remoteEndpoint.protocol == CommProtocol::COMM_PROTOCOL_ROCE) {
466 0 : auto aicpuTsRoceChannelV2 = reinterpret_cast<AicpuTsRoceChannelV2 *>(hostChannelHandles[index]);
467 0 : CHK_PRT(aicpuTsRoceChannelV2->H2DResPack(hostPackBuffers[index]));
468 : } else {
469 0 : auto aicpuTsUrmaChannel = reinterpret_cast<AicpuTsUrmaChannel *>(hostChannelHandles[index]);
470 0 : CHK_PRT(aicpuTsUrmaChannel->H2DResPack(hostPackBuffers[index]));
471 : }
472 0 : totalListNum += hostPackBuffers[index].size();
473 0 : channelSizeVec.push_back(hostPackBuffers[index].size());
474 : }
475 0 : HCCL_INFO("[%s] totalListNum[%llu]", __func__, totalListNum);
476 :
477 : // 分配连续的host内存，将序列化的地址放入其中
478 0 : hccl::HostMem hostPackBuf = hccl::HostMem::alloc(totalListNum);
479 0 : CHK_PTR_NULL(hostPackBuf.ptr());
480 0 : CHK_RET(CombineHostMemory(hostPackBuffers, hostPackBuf));
481 0 : hccl::DeviceMem devicePackBuf = hccl::DeviceMem::alloc(totalListNum);
482 0 : CHK_PTR_NULL(devicePackBuf.ptr());
483 :
484 : // 将host侧序列化内容拷贝到device侧内存中
485 0 : CHK_RET(hrtMemSyncCopy(devicePackBuf.ptr(),
486 : totalListNum,
487 : hostPackBuf.ptr(),
488 : totalListNum,
489 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
490 :
491 0 : hccl::DeviceMem channelSizeAddr = hccl::DeviceMem::alloc(channelSizeVec.size() * sizeof(u32));
492 0 : CHK_PTR_NULL(channelSizeAddr.ptr());
493 :
494 0 : CHK_RET(hrtMemSyncCopy(channelSizeAddr.ptr(),
495 : channelSizeVec.size() * sizeof(u32),
496 : channelSizeVec.data(),
497 : channelSizeVec.size() * sizeof(u32),
498 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
499 : // 为device侧的channelList分配内存
500 0 : hccl::DeviceMem deviceChannelList = hccl::DeviceMem::alloc(listNum * sizeof(ChannelHandle));
501 0 : CHK_PTR_NULL(deviceChannelList.ptr());
502 :
503 : // 填充channelParam参数
504 0 : CHK_RET(FillChannelParam(channelParam, commTag, deviceChannelList, devicePackBuf,
505 : listNum, totalListNum, channelSizeAddr));
506 :
507 : // ctx模式：检测channel是否预分配了ctx，复用deviceChannelList填ctx指针，跳过D2H
508 0 : bool isCtxMode = false;
509 0 : CHK_RET(AicpuTsChannelHelper::TryFillCtxList(hostChannelHandles, listNum, deviceChannelList, channelParam.ctxList, isCtxMode));
510 :
511 : // profiling信息
512 0 : hccl::DeviceMem remoteRankList = hccl::DeviceMem::alloc(listNum * sizeof(u32));
513 0 : CHK_PTR_NULL(remoteRankList.ptr());
514 0 : std::vector<u32> remoteRankIdList(listNum);
515 : // 集合通信场景才能开启
516 0 : if (needProfiling) {
517 0 : for ( u32 i = 0; i < listNum; ++i) {
518 0 : CHK_RET(hccl::HcclCommDfx::GetChannelRemoteRankId(commTag, hostChannelHandles[i], remoteRankIdList[i]));
519 : }
520 : // 通过安全的内存拷贝将主机内存数据传输到设备内存
521 0 : CHK_RET(hrtMemSyncCopy(remoteRankList.ptr(), listNum * sizeof(u32), remoteRankIdList.data(),
522 : listNum * sizeof(u32), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
523 0 : channelParam.remoteRankList = static_cast<u32 *>(remoteRankList.ptr());
524 : }
525 :
526 : // 调用抽离的通用内核启动函数
527 0 : CHK_RET(LaunchKernel(channelParam, binHandle, kernelName));
528 :
529 0 : if (!isCtxMode) {
530 : // 将device侧的channelList拷贝回host侧的channelList
531 0 : CHK_RET(hrtMemSyncCopy(channelHandles,
532 : listNum * sizeof(ChannelHandle),
533 : deviceChannelList.ptr(),
534 : listNum * sizeof(ChannelHandle),
535 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
536 :
537 0 : CHK_RET(FillChannelD2HMap(channelHandles, hostChannelHandles, listNum));
538 : }
539 :
540 0 : HCCL_INFO("[%s] channel kernel launch success.", __func__);
541 0 : return HCCL_SUCCESS;
542 0 : }
543 :
544 0 : HcclResult ChannelProcess::ChannelKernelLaunchForComm(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,
545 : HcommChannelDesc* hcommDesc, uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle)
546 : {
547 0 : return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, hcommDesc, listNum,
548 0 : commTag, binHandle, "RunAicpuIndOpChannelInitV2", true);
549 : }
550 :
551 1 : HcclResult ChannelProcess::ChannelKernelLaunchForBase(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,
552 : HcommChannelDesc* hcommDesc, uint32_t listNum, aclrtBinHandle binHandle)
553 : {
554 5 : return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, hcommDesc, listNum, "",
555 2 : binHandle, "RunAicpuChannelInitV2", false);
556 : }
557 :
558 : namespace {
559 :
560 : struct KHost {
561 : std::vector<std::shared_ptr<hccl::DeviceMem>> mem;
562 : std::vector<void *> ptr;
563 : std::vector<u64> sz;
564 : std::vector<u32> kind;
565 : };
566 :
567 : struct KDev {
568 : hccl::DeviceMem data;
569 : hccl::DeviceMem size;
570 : hccl::DeviceMem type;
571 : };
572 :
573 0 : HcclResult PackHost(ChannelHandle *host, uint32_t n, HcommChannelKind k, KHost &out)
574 : {
575 0 : out.mem.resize(n);
576 0 : out.ptr.resize(n);
577 0 : out.sz.resize(n);
578 0 : out.kind.resize(n);
579 0 : for (uint32_t i = 0; i < n; ++i) {
580 0 : auto *channel = reinterpret_cast<Channel *>(host[i]);
581 0 : CHK_PTR_NULL(channel);
582 0 : CHK_RET(channel->Serialize(out.mem[i]));
583 0 : CHK_PTR_NULL(out.mem[i]);
584 0 : CHK_PTR_NULL(out.mem[i]->ptr());
585 0 : out.ptr[i] = out.mem[i]->ptr();
586 0 : out.sz[i] = out.mem[i]->size();
587 0 : out.kind[i] = static_cast<u32>(k);
588 : }
589 0 : return HCCL_SUCCESS;
590 : }
591 :
592 0 : HcclResult PushDev(uint32_t n, const KHost &in, KDev &out)
593 : {
594 0 : out.data = hccl::DeviceMem::alloc(n * sizeof(void *));
595 0 : CHK_PTR_NULL(out.data.ptr());
596 0 : out.size = hccl::DeviceMem::alloc(n * sizeof(u64));
597 0 : CHK_PTR_NULL(out.size.ptr());
598 0 : out.type = hccl::DeviceMem::alloc(n * sizeof(u32));
599 0 : CHK_PTR_NULL(out.type.ptr());
600 :
601 0 : CHK_RET(hrtMemSyncCopy(out.data.ptr(), n * sizeof(void *), in.ptr.data(), n * sizeof(void *),
602 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
603 0 : CHK_RET(hrtMemSyncCopy(out.size.ptr(), n * sizeof(u64), in.sz.data(), n * sizeof(u64),
604 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
605 0 : CHK_RET(hrtMemSyncCopy(out.type.ptr(), n * sizeof(u32), in.kind.data(), n * sizeof(u32),
606 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
607 0 : return HCCL_SUCCESS;
608 : }
609 :
610 : } // namespace
611 :
612 0 : HcclResult ChannelProcess::LaunchCommonChannelKernel(ChannelHandle *channelHandles,
613 : ChannelHandle *hostChannelHandles, uint32_t listNum, HcommChannelKind channelKind, aclrtBinHandle binHandle)
614 : {
615 0 : HCCL_RUN_INFO("[%s] listNum[%u] HcommChannelRes path", __func__, listNum);
616 0 : CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", __func__, listNum), HCCL_E_PARA);
617 :
618 0 : KHost host;
619 0 : CHK_RET(PackHost(hostChannelHandles, listNum, channelKind, host));
620 0 : KDev dev;
621 0 : CHK_RET(PushDev(listNum, host, dev));
622 :
623 0 : hccl::DeviceMem deviceChannelList = hccl::DeviceMem::alloc(listNum * sizeof(ChannelHandle));
624 0 : CHK_PTR_NULL(deviceChannelList.ptr());
625 :
626 0 : HcommChannelRes channelParam{};
627 0 : channelParam.channelList = static_cast<void *>(deviceChannelList.ptr());
628 0 : channelParam.listNum = listNum;
629 0 : channelParam.channelDataListAddr = static_cast<void *>(dev.data.ptr());
630 0 : channelParam.channelDataSizeListAddr = static_cast<void *>(dev.size.ptr());
631 0 : channelParam.channelTypeListAddr = static_cast<void *>(dev.type.ptr());
632 0 : CHK_RET(hrtGetDevice(&channelParam.deviceInfo.deviceLogicId));
633 0 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(channelParam.deviceInfo.deviceLogicId), channelParam.deviceInfo.devicePhyId));
634 : DevType devType;
635 0 : CHK_RET(hrtGetDeviceType(devType));
636 0 : channelParam.deviceInfo.deviceType = static_cast<u32>(devType);
637 :
638 : // ctx模式：检测channel是否预分配了ctx，复用deviceChannelList填ctx指针，跳过D2H
639 0 : bool isCtxMode = false;
640 0 : CHK_RET(AicpuTsChannelHelper::TryFillCtxList(hostChannelHandles, listNum, deviceChannelList, channelParam.ctxList, isCtxMode));
641 :
642 0 : CHK_RET(LaunchKernelDeviceParam(channelParam, binHandle, "RunAicpuChannelInitV3"));
643 :
644 0 : if (!isCtxMode) {
645 0 : CHK_RET(hrtMemSyncCopy(channelHandles,
646 : listNum * sizeof(ChannelHandle),
647 : deviceChannelList.ptr(),
648 : listNum * sizeof(ChannelHandle),
649 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
650 :
651 0 : CHK_RET(FillChannelD2HMap(channelHandles, hostChannelHandles, listNum));
652 : }
653 0 : HCCL_INFO("[%s] channel kernel (HcommChannelRes) launch success.", __func__);
654 0 : return HCCL_SUCCESS;
655 0 : }
656 :
657 1 : HcclResult ChannelProcess::LaunchChannelKernel(ChannelHandle *channelHandles,
658 : ChannelHandle *hostChannelHandles, HcommChannelDesc *hcommDesc, uint32_t listNum, aclrtBinHandle binHandle)
659 : {
660 1 : HCCL_RUN_INFO("[%s] listNum[%u]", __func__, listNum);
661 1 : CHK_PRT_RET(listNum == 0U, HCCL_ERROR("[%s] listNum is 0", __func__), HCCL_E_PARA);
662 1 : auto *ch = reinterpret_cast<Channel *>(hostChannelHandles[0]);
663 1 : CHK_PTR_NULL(ch);
664 1 : if (ch->GetChannelKind() == HcommChannelKind::AICPU_TS_URMA
665 1 : || ch->GetChannelKind() == HcommChannelKind::AICPU_TS_UBOE
666 2 : || ch->GetChannelKind() == HcommChannelKind::AICPU_TS_UBG) {
667 1 : return ChannelKernelLaunchForBase(channelHandles, hostChannelHandles, hcommDesc, listNum, binHandle);
668 : }
669 0 : return LaunchCommonChannelKernel(channelHandles, hostChannelHandles, listNum, ch->GetChannelKind(), binHandle);
670 : }
671 :
672 22 : HcclResult ChannelProcess::PrepareUserChannels(ChannelHandle* targetChannels, ChannelHandle* userChannels,
673 : HcommChannelDesc *channelDescs, uint32_t channelNum, CommEngine engine)
674 : {
675 22 : CHK_PTR_NULL(targetChannels);
676 21 : CHK_PTR_NULL(userChannels);
677 21 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
678 :
679 20 : HCCL_INFO("[%s] engine[%s], channelNum[%u].", __func__,
680 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), channelNum);
681 20 : if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
682 3 : CHK_RET(AicpuTsChannelHelper::PreAllocChannels(targetChannels, userChannels, channelDescs, channelNum));
683 19 : } else if (engine == COMM_ENGINE_AIV) {
684 0 : CHK_RET(AivChannelHelper::PreAllocChannels(targetChannels, userChannels, channelDescs, channelNum));
685 : } else {
686 17 : HCCL_INFO("[%s] engine[%s] no need to pre-alloc.", __func__,
687 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str());
688 34 : for (uint32_t i = 0; i < channelNum; i++) {
689 17 : userChannels[i] = targetChannels[i];
690 : }
691 : }
692 19 : return HCCL_SUCCESS;
693 : }
694 :
695 0 : HcclResult ChannelProcess::ChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t *notifyNum)
696 : {
697 0 : return WithChannelByHandleLocked(channelHandle, [¬ifyNum](Channel &channel) -> HcclResult {
698 : // 锁内调用，避免 destroy 并发释放
699 0 : channel.GetNotifyNum(notifyNum);
700 0 : return HcclResult::HCCL_SUCCESS;
701 0 : });
702 : }
703 :
704 3 : HcclResult ChannelProcess::ChannelGetRemoteMems(ChannelHandle channelHandle, uint32_t *memNum, CommMem **remoteMem, char ***memInfos)
705 : {
706 3 : CHK_PTR_NULL(remoteMem);
707 2 : CHK_PTR_NULL(memNum);
708 1 : CHK_PTR_NULL(memInfos);
709 :
710 0 : return WithChannelByHandleLocked(channelHandle, [&memNum, &remoteMem, &memInfos](Channel &channel) -> HcclResult {
711 : // 锁内调用，避免 destroy 并发释放
712 0 : return channel.GetRemoteMems(memNum, remoteMem, memInfos);
713 0 : });
714 : }
715 :
716 61 : HcclResult ChannelProcess::ChannelGet(const ChannelHandle channelHandle, void **channel)
717 : {
718 61 : CHK_PTR_NULL(channel);
719 61 : int32_t deviceId = 0;
720 61 : CHK_RET(hrtGetDevice(&deviceId));
721 :
722 61 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
723 61 : DeviceChannelKey key{deviceId, channelHandle};
724 61 : const auto &D2HhandleIter = g_ChannelD2HMap.find(key);
725 61 : if (D2HhandleIter == g_ChannelD2HMap.end()) {
726 0 : HCCL_ERROR("[ChannelProcess][%s] deviceId[%d], channel[%llx] not found.", __func__, deviceId, channelHandle);
727 0 : return HcclResult::HCCL_E_NOT_FOUND;
728 : }
729 :
730 61 : const auto handle = D2HhandleIter->second;
731 61 : const auto &handleIter = g_ChannelMap.find(handle);
732 61 : if (handleIter == g_ChannelMap.end()) {
733 0 : HCCL_ERROR("[ChannelProcess][%s] deviceId[%d], channel[%llx] not found.", __func__, deviceId, handle);
734 0 : return HcclResult::HCCL_E_NOT_FOUND;
735 : }
736 61 : *channel = reinterpret_cast<void*>(handleIter->second.get());
737 61 : return HcclResult::HCCL_SUCCESS;
738 61 : }
739 :
740 0 : HcclResult ChannelProcess::ChannelKernelDestroy(ChannelHandle *channelHandles, uint32_t listNum, aclrtBinHandle binHandle)
741 : {
742 0 : HCCL_RUN_INFO("[%s] listNum[%u]", __func__, listNum);
743 0 : HcclChannelUrmaRes channelParam{};
744 0 : CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam)));
745 :
746 : // 将 host 侧的 channel handles 拷贝到 device 内存，供内核使用
747 0 : hccl::DeviceMem deviceChannelList = hccl::DeviceMem::alloc(listNum * sizeof(ChannelHandle));
748 0 : CHK_PTR_NULL(deviceChannelList.ptr());
749 0 : CHK_RET(hrtMemSyncCopy(deviceChannelList.ptr(),
750 : listNum * sizeof(ChannelHandle),
751 : channelHandles,
752 : listNum * sizeof(ChannelHandle),
753 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
754 :
755 : // 填充 channelParam（只需 channelList 和 listNum）
756 0 : channelParam.channelList = static_cast<void *>(deviceChannelList.ptr());
757 0 : channelParam.listNum = listNum;
758 :
759 : // 下 kernel
760 0 : std::string kernelName = "RunAicpuChannelDestroyV2";
761 :
762 : // 调用抽离的通用内核启动函数
763 0 : CHK_RET(LaunchKernel(channelParam, binHandle, kernelName));
764 :
765 0 : HCCL_INFO("[%s] channel kernel destroy success.", __func__);
766 0 : return HCCL_SUCCESS;
767 0 : }
768 :
769 23 : HcclResult ChannelProcess::RemoveSingleChannel(int32_t deviceId, ChannelHandle inHandle,
770 : std::vector<ChannelHandle> &deviceHandles)
771 : {
772 23 : DeviceChannelKey key{deviceId, inHandle};
773 23 : auto itH = g_ChannelD2HMap.find(key);
774 23 : if (itH == g_ChannelD2HMap.end()) {
775 6 : HCCL_ERROR("[Hcomm][%s] failed to find handle mapping in g_ChannelD2HMap, deviceId[%d], inHandle[0x%llx].",
776 : __func__, deviceId, inHandle);
777 6 : return HcclResult::HCCL_E_NOT_FOUND;
778 : }
779 17 : const ChannelHandle mappedHandle = itH->second;
780 :
781 17 : auto itC = g_ChannelMap.find(mappedHandle);
782 17 : if (itC == g_ChannelMap.end()) {
783 0 : HCCL_ERROR("[Hcomm][%s] failed to find channel in g_ChannelMap, deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx].",
784 : __func__, deviceId, inHandle, mappedHandle);
785 0 : return HcclResult::HCCL_E_NOT_FOUND;
786 : }
787 17 : deviceHandles.push_back(inHandle);
788 :
789 17 : HCCL_INFO("[Hcomm][%s] erase channel: deviceId[%d], inHandle[0x%llx], mappedHandle[0x%llx], ptr[%p]",
790 : __func__, deviceId, inHandle, mappedHandle, itC->second.get());
791 :
792 17 : g_ChannelMap.erase(itC);
793 :
794 38 : for (auto it = g_ChannelD2HMap.begin(); it != g_ChannelD2HMap.end();) {
795 21 : if (it->first.deviceId == deviceId && it->second == mappedHandle) {
796 19 : it = g_ChannelD2HMap.erase(it);
797 : } else {
798 2 : ++it;
799 : }
800 : }
801 17 : return HCCL_SUCCESS;
802 : }
803 :
804 23 : HcclResult ChannelProcess::ChannelDestroy(const ChannelHandle *channels, uint32_t channelNum, aclrtBinHandle binHandle)
805 : {
806 23 : CHK_PTR_NULL(channels);
807 23 : CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s] Invalid channelNum[0]", __func__), HCCL_E_PARA);
808 23 : HCCL_INFO("[%s] START. channelNum[%u].", __func__, channelNum);
809 :
810 23 : int32_t deviceId = 0;
811 23 : CHK_RET(hrtGetDevice(&deviceId));
812 :
813 23 : std::vector<ChannelHandle> deviceHandles;
814 :
815 : {
816 23 : std::lock_guard<std::mutex> lock(g_ChannelMapMtx);
817 40 : for (uint32_t i = 0; i < channelNum; ++i) {
818 23 : HcclResult ret = RemoveSingleChannel(deviceId, channels[i], deviceHandles);
819 23 : if (ret != HCCL_SUCCESS) {
820 6 : return ret;
821 : }
822 : }
823 23 : }
824 :
825 17 : if (!deviceHandles.empty() && binHandle) {
826 0 : CHK_RET(ChannelKernelDestroy(deviceHandles.data(), deviceHandles.size(), binHandle));
827 : }
828 17 : HCCL_INFO("[%s] SUCCESS.", __func__);
829 17 : return HCCL_SUCCESS;
830 23 : }
831 :
832 1 : HcclResult ChannelProcess::ChannelClean(const ChannelHandle *channelList, uint32_t channelNum)
833 : {
834 1 : CHK_PTR_NULL(channelList);
835 :
836 0 : for (uint32_t i = 0; i < channelNum; ++i) {
837 0 : const ChannelHandle inHandle = channelList[i];
838 : // 单锁：D2H 映射 + 查 map + 锁内调用 Clean()
839 0 : HcclResult ret = WithChannelByHandleLocked(inHandle, [](Channel &channel) -> HcclResult {
840 0 : return channel.Clean();
841 : });
842 0 : if (ret != HcclResult::HCCL_SUCCESS) {
843 0 : HCCL_ERROR("[%s] ChannelHandle Clean failed, ret = 0x%016llx, i = %u", __func__, HCCL_ERROR_CODE(ret), i);
844 0 : return ret;
845 : }
846 : }
847 :
848 0 : return HcclResult::HCCL_SUCCESS;
849 : }
850 :
851 1 : HcclResult ChannelProcess::ChannelResumeConcurrency(const ChannelHandle *channelList, uint32_t channelNum)
852 : {
853 1 : for (uint32_t i = 0; i < channelNum; ++i) {
854 0 : const ChannelHandle inHandle = channelList[i];
855 0 : HcclResult ret = WithChannelByHandleLocked(inHandle, [](Channel &channel) -> HcclResult {
856 0 : return channel.Resume();
857 : });
858 0 : if (ret != HcclResult::HCCL_SUCCESS) {
859 0 : HCCL_ERROR("[%s] Get ChannelHandle failed, ret = 0x%016llx, i = %u", __func__, HCCL_ERROR_CODE(ret), i);
860 0 : return ret;
861 : }
862 : }
863 1 : return HCCL_SUCCESS;
864 : }
865 2 : HcclResult ChannelProcess::ChannelResume(const ChannelHandle *channelList, uint32_t channelNum)
866 : {
867 2 : CHK_PTR_NULL(channelList);
868 :
869 : // 1.resume resource
870 1 : HcclResult ret = ChannelResumeConcurrency(channelList, channelNum);
871 1 : if (ret != HcclResult::HCCL_SUCCESS) {
872 1 : HCCL_ERROR("HcommChannelResumeConcurrency error, ret = 0x%016llx", HCCL_ERROR_CODE(ret));
873 1 : return ret;
874 : }
875 :
876 0 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
877 0 : auto startTime = std::chrono::steady_clock::now();
878 0 : HCCL_INFO("[%s] start resuming channels, timeout[%u]sec", __func__, timeout);
879 :
880 0 : std::vector<int32_t> statusVec(channelNum, 0);
881 0 : int32_t* statusList = statusVec.data();
882 0 : uint32_t retryCount{0};
883 : while (true) {
884 0 : HcclResult ret = ChannelGetStatus(channelList, channelNum, statusList);
885 : // 1. 检查超时
886 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
887 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
888 0 : std::chrono::steady_clock::now() - startTime).count();
889 0 : HCCL_ERROR("[%s] channel resume timeout after %u sec, channelNum[%u], elapsed[%lld]ms, retryCount[%u]",
890 : __func__, timeout, channelNum, elapsed, retryCount);
891 0 : return HCCL_E_TIMEOUT;
892 : }
893 :
894 : // 2. 处理重试（去除频繁的重试日志，一秒可能重试上千次）
895 0 : if (ret == HCCL_E_AGAIN) {
896 0 : ++retryCount;
897 0 : continue;
898 : }
899 :
900 : // 3. 处理失败
901 0 : if (ret != HCCL_SUCCESS) {
902 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
903 0 : std::chrono::steady_clock::now() - startTime).count();
904 0 : HCCL_ERROR("[%s] channel connect failed, channelNum[%u], ret[%d], elapsed[%lld]ms, retryCount[%u]",
905 : __func__, channelNum, ret, elapsed, retryCount);
906 0 : return ret;
907 : }
908 :
909 : // 4. 正常情况：所有通道连接成功
910 0 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
911 0 : std::chrono::steady_clock::now() - startTime).count();
912 0 : HCCL_INFO("[%s] all channels connected successfully, channelNum[%u], elapsed[%lld]ms, retryCount[%u]",
913 : __func__, channelNum, elapsed, retryCount);
914 0 : break;
915 0 : }
916 :
917 0 : return HcclResult::HCCL_SUCCESS;
918 0 : }
919 :
920 0 : HcclResult ChannelProcess::ChannelUpdateKernelLaunch(ChannelHandle* deviceChannelHandles, ChannelHandle* hostChannelHandles, uint32_t listNum,
921 : const std::string &commTag, aclrtBinHandle binHandle)
922 : {
923 0 : HCCL_RUN_INFO("[%s] listNum[%u], commTag[%s]", __func__, listNum, commTag.c_str());
924 0 : std::vector<std::vector<char>> hostPackBuffers(listNum);
925 0 : HcclChannelUrmaRes channelParam{};
926 0 : CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam)));
927 :
928 : // 获取host侧序列化的地址
929 0 : uint32_t totalListNum = 0;
930 0 : std::vector<u32> channelSizeVec{};
931 0 : for (uint32_t index = 0; index < listNum; index++) {
932 0 : auto aicpuTsUrmaChannel = reinterpret_cast<AicpuTsUrmaChannel *>(hostChannelHandles[index]);
933 0 : CHK_PRT(aicpuTsUrmaChannel->H2DResPack(hostPackBuffers[index])); // todo:后续只打包connction
934 0 : totalListNum += hostPackBuffers[index].size();
935 0 : channelSizeVec.push_back(hostPackBuffers[index].size());
936 : }
937 0 : HCCL_INFO("[%s] totalListNum[%llu]", __func__, totalListNum);
938 :
939 0 : hccl::DeviceMem channelSizeAddr = hccl::DeviceMem::alloc(channelSizeVec.size() * sizeof(u32));
940 0 : CHK_PTR_NULL(channelSizeAddr.ptr());
941 :
942 0 : CHK_RET(hrtMemSyncCopy(channelSizeAddr.ptr(),
943 : channelSizeVec.size() * sizeof(u32),
944 : channelSizeVec.data(),
945 : channelSizeVec.size() * sizeof(u32),
946 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
947 :
948 : // 分配连续的host内存，将序列化的地址放入其中
949 0 : hccl::HostMem hostPackBuf = hccl::HostMem::alloc(totalListNum);
950 0 : CHK_PTR_NULL(hostPackBuf.ptr());
951 0 : CHK_RET(CombineHostMemory(hostPackBuffers, hostPackBuf));
952 0 : hccl::DeviceMem devicePackBuf = hccl::DeviceMem::alloc(totalListNum);
953 0 : CHK_PTR_NULL(devicePackBuf.ptr());
954 :
955 : // 将host侧序列化内容拷贝到device侧内存中
956 0 : CHK_RET(hrtMemSyncCopy(devicePackBuf.ptr(),
957 : totalListNum,
958 : hostPackBuf.ptr(),
959 : totalListNum,
960 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
961 :
962 : // 填充channelParam参数
963 0 : s32 sRet = strncpy_s(channelParam.hcomId, HCOMID_MAX_LENGTH, commTag.c_str(), HCOMID_MAX_LENGTH - 1);
964 0 : CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] str copy fail. return[%d]", __func__, sRet), HCCL_E_INTERNAL);
965 0 : channelParam.listNum = listNum;
966 0 : channelParam.uniqueIdAddr = static_cast<void *>(devicePackBuf.ptr());
967 0 : channelParam.uniqueIdSize = totalListNum;
968 0 : channelParam.channelSizeAddr = static_cast<void *>(channelSizeAddr.ptr());
969 :
970 : // 将 host 侧的 channel handles 拷贝到 device 内存，供内核使用
971 0 : hccl::DeviceMem deviceChannelList = hccl::DeviceMem::alloc(listNum * sizeof(ChannelHandle));
972 0 : CHK_PTR_NULL(deviceChannelList.ptr());
973 0 : CHK_RET(hrtMemSyncCopy(deviceChannelList.ptr(),
974 : listNum * sizeof(ChannelHandle),
975 : deviceChannelHandles,
976 : listNum * sizeof(ChannelHandle),
977 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
978 0 : channelParam.channelList = static_cast<void *>(deviceChannelList.ptr());
979 :
980 : // 调用抽离的通用内核启动函数
981 0 : std::string kernelName = "RunAicpuIndOpChannelUpdateV2";
982 0 : CHK_RET(LaunchKernel(channelParam, binHandle, kernelName));
983 :
984 0 : HCCL_INFO("[%s] channel kernel launch success.", __func__);
985 0 : return HCCL_SUCCESS;
986 0 : }
987 :
988 : }
|
