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 "op_base.h"
12 : #include <algorithm>
13 : #include <future>
14 : #include <map>
15 : #include <string>
16 : #include <hccl/hccl_types.h>
17 : #include "aicpu_operator_pub.h"
18 : #include "coll_alg_param.h"
19 : #include "hccl/base.h"
20 : #include "kernel_tiling/kernel_tiling.h"
21 : #include "param_check_pub.h"
22 : #include "hccl_tiling_msg.h"
23 : #include "op_base_v2.h"
24 :
25 : using namespace std;
26 : using namespace hccl;
27 :
28 : namespace {
29 : const u32 MC2_TILING_VERSION = 2U;
30 : const u32 ENABLE_AICPU_COMM_ENGINE = 0U;
31 : }
32 0 : HcclResult HcclGetInitTilingList(const void *mc2Tiling, const void *p[], uint32_t &cnt)
33 : {
34 0 : const u32 *versionPtr = static_cast<const u32 *>(mc2Tiling);
35 0 : const u32 version = *(versionPtr++);
36 0 : CHK_PRT_RET(version < MC2_TILING_VERSION, HCCL_ERROR("Invalid tiling version %u.", version), HCCL_E_PARA);
37 :
38 0 : cnt = *(versionPtr++);
39 0 : CHK_PRT_RET(cnt > MAX_HCOM_NUM, HCCL_ERROR("Invalid hcom tiling number %u.", cnt), HCCL_E_PARA);
40 :
41 0 : u64 serverCfgAddr = reinterpret_cast<u64>(versionPtr) + sizeof(Mc2ServerCfg);
42 0 : for (uint32_t i = 0U; i < cnt; ++i) {
43 0 : if (version == MC2_TILING_VERSION) {
44 0 : p[i] = reinterpret_cast<const void *>(serverCfgAddr + i * sizeof(Mc2HcommCfg));
45 : } else {
46 0 : p[i] = reinterpret_cast<const void *>(reinterpret_cast<const u8 *>(mc2Tiling) + versionPtr[i]);
47 : }
48 : }
49 0 : HCCL_INFO("HcclGetInitTilingList version[%u] cnt[%u]", version, cnt);
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult HcclMc2ComResourceByTiling(HcclComm comm, void *mc2Tiling, rtStream_t &aicpuStream)
54 : {
55 : const void *tilingList[MAX_HCOM_NUM];
56 : uint32_t tilingNum;
57 0 : CHK_RET(HcclGetInitTilingList(mc2Tiling, tilingList, tilingNum));
58 0 : CHK_PRT_RET(tilingNum == 0, HCCL_ERROR("Invalid tilingNum %u.", tilingNum), HCCL_E_PARA);
59 :
60 0 : hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
61 0 : string commIdentifier = hcclComm->GetIdentifier();
62 0 : bool isAicpuCommEngine = false;
63 0 : for (uint32_t i = 0U; i < tilingNum; ++i) {
64 0 : const HcclApi::Mc2CcTilingInner *tiling = static_cast<const HcclApi::Mc2CcTilingInner *>(tilingList[i]);
65 0 : if (tiling == nullptr || string(tiling->groupName) != commIdentifier) {
66 0 : continue;
67 : }
68 :
69 0 : OpParam opParam;
70 0 : opParam.tag = string(tiling->groupName) + to_string(tiling->opType) + string("_mc2");
71 0 : opParam.stream = Stream(aicpuStream);
72 0 : opParam.reduceType = static_cast<HcclReduceOp>(tiling->reduceType);
73 0 : opParam.syncMode = SyncMode::DEFAULT_TIMEWAITSYNCMODE;
74 0 : opParam.aicpuUnfoldMode = true;
75 0 : opParam.opType = static_cast<HcclCMDType>(tiling->opType);
76 0 : if (opParam.opType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
77 0 : opParam.BatchWriteDataDes.queueNum = LOCAL_STREAM_MAX_NUM;
78 0 : HCCL_INFO("Requiring %u queues for batch-write.", opParam.BatchWriteDataDes.queueNum);
79 : }
80 0 : HCCL_INFO("Comm resource will be created for group %s. isAicpuCommEngine[%d] commEngine[%u]",
81 : commIdentifier.c_str(), isAicpuCommEngine, tiling->commEngine);
82 0 : CHK_RET(hcclComm->AllocComResourceByTiling(tiling->algConfig, reinterpret_cast<void *>(&opParam)));
83 : // commEngine为0代表使能AICPU引擎
84 0 : if (!isAicpuCommEngine && tiling->commEngine == ENABLE_AICPU_COMM_ENGINE) {
85 0 : isAicpuCommEngine = true;
86 : }
87 0 : }
88 :
89 0 : if (isAicpuCommEngine) {
90 0 : CHK_RET(hcclComm->SetAicpuCommEngine(isAicpuCommEngine));
91 : }
92 :
93 0 : return HCCL_SUCCESS;
94 0 : }
95 :
96 1 : HcclResult HcclMc2ComOpResCtx(HcclComm comm, uint8_t opType, HcclDataType srcDataType, HcclDataType dstDataType,
97 : HcclReduceOp reduceType, uint64_t count, char *algConfig, uint32_t commEngine, rtStream_t &aicpuStream)
98 : {
99 1 : HCCL_DEBUG("HcclMc2ComOpResCtx: srcDataType[%d], dstDataType[%d], count[%llu]", srcDataType, dstDataType, count);
100 1 : hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
101 1 : string commIdentifier = hcclComm->GetIdentifier();
102 :
103 1 : OpParam opParam;
104 1 : opParam.tag = string(commIdentifier) + to_string(opType) + string("_mc2");
105 1 : opParam.stream = Stream(aicpuStream);
106 1 : opParam.reduceType = static_cast<HcclReduceOp>(reduceType);
107 1 : opParam.syncMode = SyncMode::DEFAULT_TIMEWAITSYNCMODE;
108 1 : opParam.aicpuUnfoldMode = true;
109 1 : opParam.opType = static_cast<HcclCMDType>(opType);
110 1 : if (opParam.opType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
111 0 : opParam.BatchWriteDataDes.queueNum = LOCAL_STREAM_MAX_NUM;
112 0 : HCCL_INFO("Requiring %u queues for batch-write.", opParam.BatchWriteDataDes.queueNum);
113 : }
114 1 : HCCL_INFO("Comm resource will be created for group[%s] commEngine[%u]", commIdentifier.c_str(), commEngine);
115 3 : CHK_RET(hcclComm->AllocComResourceByTiling(algConfig, reinterpret_cast<void *>(&opParam)));
116 :
117 1 : if (commEngine == COMM_ENGINE_AICPU) {
118 0 : CHK_RET(hcclComm->SetAicpuCommEngine(true));
119 : }
120 :
121 1 : return HCCL_SUCCESS;
122 1 : }
123 :
124 6 : HcclResult HcclCreateOpResCtxInner(HcclComm comm, uint8_t opType, HcclDataType srcDataType, HcclDataType dstDataType,
125 : HcclReduceOp reduceType, uint64_t count, char *algConfig, uint32_t commEngine, void **opResCtx)
126 : {
127 : // 校验
128 6 : CHK_PTR_NULL(comm);
129 5 : CHK_PTR_NULL(algConfig);
130 4 : CHK_PTR_NULL(opResCtx);
131 :
132 : DevType devType;
133 3 : CHK_RET(hrtGetDeviceType(devType));
134 3 : if ((devType != DevType::DEV_TYPE_910_93) && (devType != DevType::DEV_TYPE_910B)) {
135 1 : HCCL_ERROR("[HcclCreateOpResCtxInner] devType[%d] is not supported", devType);
136 1 : return HCCL_E_NOT_SUPPORT;
137 : }
138 :
139 2 : HcclUs startut = TIME_NOW();
140 2 : uint64_t streamMode = 0; //streamMode未使用，固定传0
141 2 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
142 2 : string commIdentifier = hcclComm->GetIdentifier();
143 2 : HCCL_INFO("[%s]commIdentifier[%s], opType[%d]", __func__, commIdentifier.c_str(), opType);
144 2 : string cclBufferName = hcclComm->GetCCLbufferName();
145 2 : bool isShareComm = cclBufferName.empty() ? false : true;
146 2 : if (isShareComm) {
147 0 : HCCL_RUN_WARNING("MC2 using share CCLbuffer[%s], potential conflict with coll communicator", cclBufferName.c_str());
148 : }
149 :
150 : // 根据streamMode创建aicpuStream
151 2 : rtStream_t aicpuStream{};
152 2 : CHK_RET(hcclComm->Mc2AiCpuStreamAllocAndGet(streamMode, aicpuStream));
153 :
154 : char stackLogBuffer[LOG_TMPBUF_SIZE];
155 2 : u32 localRank = INVALID_VALUE_RANKID;
156 2 : CHK_RET(hcclComm->GetUserRank(localRank));
157 :
158 : /* 接口交互信息日志 */
159 2 : if (GetExternalInputHcclEnableEntryLog()) {
160 0 : s32 ret = snprintf_s(stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "commIdentifier[%s]", commIdentifier.c_str());
161 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, commIdentifier[%s].", commIdentifier.c_str()));
162 :
163 0 : std::string logInfo = "MC2 create resource by tiling: localRank[" + std::to_string(localRank)
164 0 : + "]" + std::string(stackLogBuffer);
165 0 : CHK_RET(hcclComm->SaveTraceInfo(logInfo));
166 0 : }
167 :
168 2 : if (DevType::DEV_TYPE_910_93 == devType) {
169 1 : CHK_RET(HcclMc2ComOpResCtx(comm, opType, srcDataType, dstDataType, reduceType, count, algConfig, commEngine, aicpuStream));
170 :
171 : // 获取 commContext
172 1 : hcclComm->GetCommResource(*opResCtx);
173 1 : if (*opResCtx == nullptr) {
174 0 : HCCL_ERROR("[%s] GetCommResource failed, opResCtx is nullptr, commIdentifier[%s]", __func__, commIdentifier.c_str());
175 0 : return HCCL_E_INTERNAL;
176 : }
177 : } else {
178 1 : string tag = "CreatecomResource_" + commIdentifier;
179 1 : u32 moduleNum = hcclComm->GetModuleNum();
180 1 : if (moduleNum > HCCL_DEVICE_NUM_ONE) {
181 0 : tag += HCCL_MC2_MULTISERVER_SUFFIX;
182 : }
183 1 : if (commEngine == COMM_ENGINE_AICPU) {
184 1 : CHK_RET(hcclComm->SetAicpuCommEngine(true));
185 : }
186 1 : if (LIKELY(hcclComm->GetCommResource(tag, opResCtx))) {
187 0 : return HCCL_SUCCESS;
188 : }
189 3 : CHK_RET(hcclComm->CreateCommResource(tag, aicpuStream, true, opResCtx, algConfig));
190 1 : }
191 :
192 2 : if (GetExternalInputHcclEnableEntryLog()) {
193 0 : HcclUs endut = TIME_NOW();
194 : /* 关键状态记录 */
195 : std::string endInfo = "MC2 create resource take time ["
196 0 : + std::to_string(DURATION_US(endut - startut).count()) + "]us, localRank["
197 0 : + std::to_string(localRank) + "] " + std::string(stackLogBuffer);
198 0 : CHK_RET(hcclComm->SaveTraceInfo(endInfo));
199 0 : }
200 :
201 2 : return HCCL_SUCCESS;
202 2 : }
203 :
204 : #ifdef __cplusplus
205 : extern "C" {
206 : #endif
207 0 : HcclResult HcclAllocComResourceByTiling(HcclComm comm, void* stream, void* Mc2Tiling, void** commContext)
208 : {
209 : // 校验
210 0 : CHK_PTR_NULL(comm);
211 0 : CHK_PTR_NULL(stream);
212 0 : CHK_PTR_NULL(Mc2Tiling);
213 0 : CHK_PTR_NULL(commContext);
214 :
215 0 : HcclUs startut = TIME_NOW();
216 0 : uint64_t streamMode = 0; //streamMode未使用，固定传0
217 : // 兼容老版本
218 0 : uint32_t *pVersion = reinterpret_cast<uint32_t *>(Mc2Tiling);
219 : DevType devType;
220 0 : CHK_RET(hrtGetDeviceType(devType));
221 0 : HCCL_INFO("[%s]version ptr[%p] val[%u] devType[%u]", __func__, pVersion, *pVersion, devType);
222 : #if (!defined (HCCD)) && (!defined (CCL_KERNEL_AICPU))
223 0 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
224 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
225 : CHK_RET(HcclAllocComResourceByTilingV2(hcclComm->GetCommunicatorV2(), stream, Mc2Tiling, commContext));
226 : return HCCL_SUCCESS;
227 : }());
228 : #endif
229 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
230 0 : string commIdentifier = hcclComm->GetIdentifier();
231 0 : HCCL_INFO("[%s]commIdentifier[%s]", __func__, commIdentifier.c_str());
232 0 : string cclBufferName = hcclComm->GetCCLbufferName();
233 0 : bool isShareComm = cclBufferName.empty() ? false : true;
234 0 : if (isShareComm) {
235 0 : HCCL_RUN_WARNING("MC2 using share CCLbuffer[%s], potential conflict with coll communicator", cclBufferName.c_str());
236 : }
237 0 : if (*pVersion < MC2_TILING_VERSION || devType != DevType::DEV_TYPE_910_93) {
238 0 : return HcclCreateComResourceByComm(comm, streamMode, true, commContext, true, Mc2Tiling);
239 : }
240 :
241 : // 根据streamMode创建aicpuStream
242 0 : rtStream_t aicpuStream{};
243 0 : CHK_RET(hcclComm->Mc2AiCpuStreamAllocAndGet(streamMode, aicpuStream));
244 :
245 : char stackLogBuffer[LOG_TMPBUF_SIZE];
246 :
247 0 : u32 localRank = INVALID_VALUE_RANKID;
248 0 : CHK_RET(hcclComm->GetUserRank(localRank));
249 :
250 : /* 接口交互信息日志 */
251 0 : if (GetExternalInputHcclEnableEntryLog()) {
252 0 : s32 ret = snprintf_s(stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "commIdentifier[%s], version[%u]",
253 : commIdentifier.c_str(), *pVersion);
254 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, commIdentifier[%s].", commIdentifier.c_str()));
255 :
256 0 : std::string logInfo = "MC2 create resource by tiling: localRank[" + std::to_string(localRank)
257 0 : + "]" + std::string(stackLogBuffer);
258 0 : CHK_RET(hcclComm->SaveTraceInfo(logInfo));
259 0 : }
260 :
261 0 : CHK_RET(HcclMc2ComResourceByTiling(comm, Mc2Tiling, aicpuStream));
262 :
263 : // 获取 commContext
264 0 : hcclComm->GetCommResource(*commContext);
265 0 : if (*commContext == nullptr) {
266 0 : HCCL_ERROR("[%s] GetCommResource failed, commContext is nullptr, commIdentifier[%s]", __func__, commIdentifier.c_str());
267 0 : return HCCL_E_INTERNAL;
268 : }
269 :
270 0 : if (GetExternalInputHcclEnableEntryLog()) {
271 0 : HcclUs endut = TIME_NOW();
272 : /* 关键状态记录 */
273 : std::string endInfo = "MC2 create resource take time ["
274 0 : + std::to_string(DURATION_US(endut - startut).count()) + "]us, localRank["
275 0 : + std::to_string(localRank) + "] " + std::string(stackLogBuffer);
276 0 : CHK_RET(hcclComm->SaveTraceInfo(endInfo));
277 0 : }
278 :
279 0 : return HCCL_SUCCESS;
280 0 : }
281 :
282 : #if (!defined (HCCD)) && (!defined (CCL_KERNEL_AICPU))
283 0 : HcclResult HcclGetOpArgs(void **opArgs)
284 : {
285 0 : CHK_PTR_NULL(opArgs);
286 0 : HCCLV2_FUNC_RUN(HcclGetOpArgsV2(opArgs));
287 0 : return HCCL_SUCCESS;
288 : }
289 :
290 0 : HcclResult HcclFreeOpArgs(void *opArgs)
291 : {
292 0 : CHK_PTR_NULL(opArgs);
293 0 : HCCLV2_FUNC_RUN(HcclFreeOpArgsV2(opArgs));
294 0 : return HCCL_SUCCESS;
295 : }
296 :
297 0 : HcclResult HcclSetOpSrcDataType(void *opArgs, uint8_t srcDataType)
298 : {
299 0 : CHK_PTR_NULL(opArgs);
300 0 : HCCLV2_FUNC_RUN(HcclSetOpSrcDataTypeV2(opArgs, srcDataType));
301 0 : return HCCL_SUCCESS;
302 : }
303 :
304 0 : HcclResult HcclSetOpDstDataType(void *opArgs, uint8_t dstDataType)
305 : {
306 0 : CHK_PTR_NULL(opArgs);
307 0 : HCCLV2_FUNC_RUN(HcclSetOpDstDataTypeV2(opArgs, dstDataType));
308 0 : return HCCL_SUCCESS;
309 : }
310 :
311 0 : HcclResult HcclSetOpReduceType(void *opArgs, uint32_t reduceType)
312 : {
313 0 : CHK_PTR_NULL(opArgs);
314 0 : HCCLV2_FUNC_RUN(HcclSetOpReduceTypeV2(opArgs, reduceType));
315 0 : return HCCL_SUCCESS;
316 : }
317 :
318 0 : HcclResult HcclSetOpCount(void *opArgs, uint64_t count)
319 : {
320 0 : CHK_PTR_NULL(opArgs);
321 0 : HCCLV2_FUNC_RUN(HcclSetOpCountV2(opArgs, count));
322 0 : return HCCL_SUCCESS;
323 : }
324 :
325 0 : HcclResult HcclSetOpAlgConfig(void *opArgs, char *algConfig)
326 : {
327 0 : CHK_PTR_NULL(opArgs);
328 0 : CHK_PTR_NULL(algConfig);
329 0 : HCCLV2_FUNC_RUN(HcclSetOpAlgConfigV2(opArgs, algConfig));
330 0 : return HCCL_SUCCESS;
331 : }
332 :
333 0 : HcclResult HcclSetOpCommEngine(void *opArgs, uint8_t commEngine)
334 : {
335 0 : CHK_PTR_NULL(opArgs);
336 0 : HCCLV2_FUNC_RUN(HcclSetOpCommEngineV2(opArgs, commEngine));
337 0 : return HCCL_SUCCESS;
338 : }
339 :
340 0 : HcclResult HcclCommResPrepare(HcclComm comm, char *opName, void *opArgs, void **addr)
341 : {
342 0 : CHK_PTR_NULL(comm);
343 0 : CHK_PTR_NULL(opName);
344 0 : CHK_PTR_NULL(opArgs);
345 0 : CHK_PTR_NULL(addr);
346 0 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
347 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
348 : CHK_RET(HcclCommResPrepareV2(hcclComm->GetCommunicatorV2(), opName, opArgs, addr));
349 : return HCCL_SUCCESS;
350 : }());
351 0 : return HCCL_SUCCESS;
352 : }
353 :
354 0 : HcclResult HcclDevMemAcquire(HcclComm comm, const char *memTag, uint64_t *size, void **addr, bool *newCreated)
355 : {
356 0 : CHK_PTR_NULL(comm);
357 0 : CHK_PTR_NULL(size);
358 0 : CHK_PTR_NULL(addr);
359 0 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
360 0 : if (hcclComm->GetCommunicatorV2() != nullptr) {
361 0 : HCCLV2_FUNC_RUN(HcclDevMemAcquireV2(hcclComm->GetCommunicatorV2(), memTag, size, addr, newCreated));
362 : } else {
363 0 : CHK_RET(hcclComm->GetDevMemWorkSpace(memTag, size, addr, newCreated));
364 : }
365 0 : return HCCL_SUCCESS;
366 : }
367 :
368 9 : HcclResult HcclGetRemoteIpcHcclBuf(HcclComm comm, uint64_t remoteRank, void **addr, uint64_t *size)
369 : {
370 9 : CHK_PTR_NULL(comm);
371 8 : CHK_PTR_NULL(addr);
372 7 : CHK_PTR_NULL(size);
373 :
374 6 : HCCLV2_FUNC_RUN([&]() -> HcclResult {
375 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
376 : CHK_RET(HcclGetRemoteIpcHcclBuf(hcclComm->GetCommunicatorV2(), remoteRank, addr, size));
377 : return HCCL_SUCCESS;
378 : }());
379 6 : hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
380 6 : void *opResCtx = nullptr;
381 6 : hcclComm->GetCommResource(opResCtx);
382 6 : if (opResCtx == nullptr) {
383 0 : HCCL_ERROR("[%s]comm[%s] remoteRank[%llu] get resource fail", __func__, hcclComm->GetIdentifier().c_str(), remoteRank);
384 0 : return HCCL_E_PARA;
385 : }
386 :
387 6 : CHK_RET(hcclComm->GetRemoteCCLBuf(remoteRank, addr, size));
388 4 : if (*addr == nullptr) {
389 1 : u32 localRank = INVALID_VALUE_RANKID;
390 1 : CHK_RET(hcclComm->GetUserRank(localRank));
391 1 : HCCL_ERROR("[%s]comm[%s] get remote CCL buffer fail, ret is nullptr. Possible reasons:"
392 : "The selected AlgConfig has not create link between localRank[%u] to remoteRank[%llu].",
393 : __func__, hcclComm->GetIdentifier().c_str(), localRank, remoteRank);
394 1 : return HCCL_E_PTR;
395 : }
396 :
397 3 : return HCCL_SUCCESS;
398 : }
399 : #endif
400 :
401 : #ifdef __cplusplus
402 : }
403 : #endif
|
