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