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