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