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 <numeric>
12 : #include <string>
13 : #include <dlog_pub.h>
14 : #include <hccl/hccl_types.h>
15 : #include "log.h"
16 : #include "securec.h"
17 : #include "aicpu_communicator.h"
18 : #include "common/aicpu_hccl_common.h"
19 : #include "common/aicpu_sqe_context.h"
20 : #include "executor_tracer.h"
21 : #include "utils/hccl_aicpu_utils.h"
22 : #include "hccl_types.h"
23 : #include "framework/aicpu_hccl_process.h"
24 : #include "dtype_common.h"
25 : #include "aicpu_one_side_service.h"
26 : #include "coll_comm_aicpu_mgr.h"
27 :
28 : using namespace hccl;
29 : using namespace HcclApi;
30 : namespace {
31 : AicpuComContext g_comContext[CLUSTER_CNT];
32 : DevType g_devType = DevType::DEV_TYPE_COUNT;
33 : } // namespace
34 :
35 0 : DevType AicpuHcclProcess::AicpuGetInnerDevType() { return g_devType; }
36 :
37 : static constexpr uint32_t ALLTOALLV_INFO_INDEX_2 = 2;
38 : static constexpr uint32_t ALLTOALLV_INFO_INDEX_3 = 3;
39 : static constexpr uint32_t ALLTOALLV_INFO_INDEX_4 = 4;
40 :
41 24367 : AicpuComContext* AicpuGetComContext()
42 : {
43 24367 : auto clusterId = HcclAicpuUtils::GetCurClusterId();
44 24367 : return &g_comContext[clusterId];
45 : }
46 :
47 229257849 : void AicpuGetAllComContext(AicpuComContext*& contextBase, uint32_t& contextNum)
48 : {
49 229257849 : contextBase = &g_comContext[0];
50 229257849 : contextNum = CLUSTER_CNT;
51 229257849 : return;
52 : }
53 :
54 140 : void AicpuHcclProcess::CallMC2MaintenanceThread(AicpuComContext* ctx)
55 : {
56 140 : if (!IsSupportStartMC2MaintenanceThread()) {
57 0 : return;
58 : }
59 :
60 140 : if (ctx->commOpenStatus && (ctx->devType == DevType::DEV_TYPE_310P1 || ctx->devType == DevType::DEV_TYPE_310P3)) {
61 0 : return;
62 : }
63 :
64 140 : HCCL_INFO(
65 : "Register back ground func on device type %u, status %u.", static_cast<u32>(ctx->devType),
66 : static_cast<u32>(ctx->commOpenStatus));
67 140 : hrtHalStartMC2MaintenanceThread(
68 : dfx_tracer::ExecutorTracer::BackGroundDfx, ctx, dfx_tracer::ExecutorTracer::StopBackGroundDfx, ctx);
69 : }
70 :
71 4 : HcclResult AicpuHcclProcess::CalcDataSize(
72 : HcclCMDType op, HcclDataType type, u64 count, u32 rankSize, u64& inputSize, u64& outputSize)
73 : {
74 4 : u32 perDataSize = DataUnitSize(type);
75 4 : if (perDataSize == 0) {
76 1 : HCCL_ERROR("[AicpuHcclProcess][CalcDataSize] type [%u] DataUnitSize is 0", type);
77 1 : return HCCL_E_PARA;
78 : }
79 :
80 3 : switch (op) {
81 1 : case HcclCMDType::HCCL_CMD_ALLGATHER:
82 1 : inputSize = count * perDataSize;
83 1 : outputSize = rankSize * count * perDataSize;
84 1 : break;
85 1 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER:
86 : case HcclCMDType::HCCL_CMD_SCATTER:
87 1 : inputSize = rankSize * count * perDataSize;
88 1 : outputSize = count * perDataSize;
89 1 : break;
90 1 : case HcclCMDType::HCCL_CMD_ALLREDUCE:
91 : case HcclCMDType::HCCL_CMD_BROADCAST:
92 : case HcclCMDType::HCCL_CMD_REDUCE:
93 : case HcclCMDType::HCCL_CMD_SEND:
94 : case HcclCMDType::HCCL_CMD_RECEIVE:
95 : default:
96 1 : inputSize = count * perDataSize;
97 1 : outputSize = count * perDataSize;
98 1 : break;
99 : }
100 :
101 3 : HCCL_DEBUG(
102 : "[AicpuHcclProcess][CalcDataSize] perDataSize %u count %lu rankSize %u input %lu output %lu", perDataSize,
103 : count, rankSize, inputSize, outputSize);
104 3 : return HCCL_SUCCESS;
105 : }
106 :
107 0 : HcclResult AicpuHcclProcess::CalcDataSizeV(hccl::OpParam& param, u32 rankSize)
108 : {
109 0 : const HcclDataType type = param.GetDataType();
110 0 : u32 perDataSize = DataUnitSize(type);
111 0 : if (perDataSize == 0) {
112 0 : HCCL_ERROR("[AicpuHcclProcess][CalcDataSizeV] type[%u] DataUnitSize is 0", type);
113 0 : return HCCL_E_PARA;
114 : }
115 :
116 0 : const u32 rankId = param.srcRank;
117 0 : if (rankId >= rankSize) {
118 0 : HCCL_ERROR(
119 : "[AicpuHcclProcess][CalcDataSizeV] rankId[%u] should not be bigger than rankSize[%u]", rankId, rankSize);
120 0 : return HCCL_E_PARA;
121 : }
122 :
123 0 : const HcclCMDType op = param.opType;
124 0 : switch (op) {
125 0 : case HcclCMDType::HCCL_CMD_ALLGATHER_V: {
126 0 : const u64* counts = static_cast<u64*>(param.VDataDes.counts);
127 0 : const u64 totalSize = std::accumulate(counts, counts + rankSize, 0ULL) * perDataSize;
128 0 : param.inputSize = param.GetDataCount(rankId) * perDataSize;
129 0 : param.outputSize = totalSize;
130 0 : break;
131 : }
132 0 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V: {
133 0 : const u64* counts = static_cast<u64*>(param.VDataDes.counts);
134 0 : const u64 totalSize = std::accumulate(counts, counts + rankSize, 0ULL) * perDataSize;
135 0 : param.inputSize = totalSize;
136 0 : param.outputSize = param.GetDataCount(rankId) * perDataSize;
137 0 : break;
138 : }
139 0 : default:
140 0 : HCCL_ERROR("[AicpuHcclProcess][CalcDataSizeV] op[%u] is not supported", op);
141 0 : return HCCL_E_PARA;
142 : }
143 :
144 0 : HCCL_DEBUG(
145 : "[AicpuHcclProcess][CalcDataSizeV] opType[%u] perDataSize[%u] rankId[%u] rankSize[%u] input[%llu] "
146 : "output[%llu]",
147 : op, perDataSize, rankId, rankSize, param.inputSize, param.outputSize);
148 0 : return HCCL_SUCCESS;
149 : }
150 :
151 0 : u64 AicpuHcclProcess::CalcOpTilingVDataDesVDataLen(u32 rankSize)
152 : {
153 0 : const u32 vFactor = 2; // counts和displs 2个变长数组
154 0 : return vFactor * rankSize * sizeof(u64);
155 : }
156 :
157 13 : u32 AicpuHcclProcess::AicpuRpcResInitV2(HcclOpResParam* commParam, bool isCustom)
158 : {
159 13 : HCCL_DEBUG("[AicpuHcclProcess][AicpuRpcResInitV2]Entry AicpuRpcResInitV2 process-------");
160 13 : hccl::HcclCommAicpu* commAicpu = nullptr;
161 13 : HcclResult ret = HCCL_SUCCESS;
162 13 : std::string group = commParam->hcomId;
163 13 : CHK_RET(AcquireAicpuComm(group, &commAicpu));
164 13 : if (commAicpu == nullptr) {
165 0 : HCCL_ERROR("[AicpuHcclProcess][AicpuRpcResInitV2]commAicpu is null group[%s]", group.c_str());
166 0 : return 1U;
167 : }
168 13 : ret = commAicpu->Init(commParam, isCustom);
169 13 : CHK_PRT_RET(
170 : ret != HCCL_SUCCESS,
171 : HCCL_ERROR(
172 : "[AicpuHcclProcess][AicpuRpcResInitV2]errNo[0x%016llx] Failed to init comm resource group[%s]",
173 : HCCL_ERROR_CODE(ret), group.c_str()),
174 : ret);
175 13 : HCCL_DEBUG("[AicpuHcclProcess][AicpuRpcResInitV2]AicpuRpcResInitV2 process end-------");
176 13 : if (g_devType == DevType::DEV_TYPE_COUNT) {
177 1 : CHK_RET(hrtHalGetDeviceType(commAicpu->GetDevId(), g_devType));
178 1 : HCCL_RUN_INFO(
179 : "[AicpuHcclProcess][AicpuRpcResInitV2] devId %u, get devType %u", commAicpu->GetDevId(),
180 : static_cast<u32>(g_devType));
181 : }
182 13 : HCCL_INFO(
183 : "[AicpuHcclProcess][AicpuRpcResInitV2] get PlatformVersion %u, %u", static_cast<u32>(g_devType),
184 : commAicpu->GetDevId());
185 13 : AicpuComContext* ctx = AicpuGetComContext();
186 13 : CallMC2MaintenanceThread(ctx);
187 :
188 13 : return 0;
189 13 : }
190 :
191 13 : HcclResult AicpuHcclProcess::AcquireAicpuComm(const std::string& group, HcclCommAicpu** aicpuCommPtr)
192 : {
193 : // 通过 CollCommAicpuMgr 单例管理通信域，init 路径不标记使用中
194 13 : CollCommAicpu* collComm = nullptr;
195 13 : CHK_RET(CollCommAicpuMgr::GetInstance().AcquireAndCreateComm(group, &collComm));
196 13 : if (collComm == nullptr) {
197 0 : HCCL_ERROR("[%s]Failed to acquire comm group[%s]", __func__, group.c_str());
198 0 : return HCCL_E_PTR;
199 : }
200 :
201 13 : HcclCommAicpu* legacyComm = collComm->GetLegacy910CollComm();
202 13 : if (legacyComm == nullptr) {
203 11 : std::shared_ptr<HcclCommAicpu> newComm;
204 : try {
205 11 : newComm = std::make_shared<HcclCommAicpu>();
206 0 : } catch (std::exception& e) {
207 0 : HCCL_ERROR("[%s]Failed, exception caught:%s", __func__, e.what());
208 0 : return HCCL_E_PTR;
209 0 : }
210 11 : if (UNLIKELY(!newComm)) {
211 0 : HCCL_ERROR("[%s]errNo[0x%016llx] aicpuComm is nullptr", __func__, HCCL_ERROR_CODE(HCCL_E_PTR));
212 0 : return HCCL_E_PTR;
213 : }
214 11 : collComm->SetLegacy910CollComm(newComm);
215 11 : legacyComm = newComm.get();
216 11 : HCCL_INFO("[%s]Created new legacy comm group [%s]", __func__, group.c_str());
217 11 : } else {
218 2 : HCCL_INFO("[%s]Reuse existing legacy comm group [%s]", __func__, group.c_str());
219 : }
220 :
221 13 : *aicpuCommPtr = legacyComm;
222 13 : return HCCL_SUCCESS;
223 : }
224 :
225 0 : HcclResult AicpuHcclProcess::AicpuIndOpCommInit(CommAicpuParam* commAicpuParam)
226 : {
227 0 : hccl::HcclCommAicpu* commAicpu = nullptr;
228 0 : HcclResult ret = HCCL_SUCCESS;
229 0 : std::string group = commAicpuParam->hcomId;
230 0 : CHK_RET(AcquireAicpuComm(group, &commAicpu));
231 0 : if (commAicpu == nullptr) {
232 0 : HCCL_ERROR("[AicpuHcclProcess][AicpuIndOpCommInit]commAicpu is null group[%s]", group.c_str());
233 0 : return HCCL_E_PTR;
234 : }
235 0 : ret = commAicpu->InitAicpuIndOp(commAicpuParam);
236 0 : CHK_PRT_RET(
237 : ret != HCCL_SUCCESS,
238 : HCCL_ERROR(
239 : "[AicpuHcclProcess][AicpuIndOpCommInit]errNo[0x%016llx] Failed to init independent op comm group[%s]",
240 : HCCL_ERROR_CODE(ret), group.c_str()),
241 : ret);
242 0 : return HCCL_SUCCESS;
243 0 : }
244 :
245 0 : HcclResult AicpuHcclProcess::AicpuRegOpInfo(void* opInfo, u32 size)
246 : {
247 0 : CollCommAicpu* collComm = CollCommAicpuMgr::GetInstance().GetCurrentComm();
248 0 : CHK_PTR_NULL(collComm);
249 0 : HcclCommAicpu* legacyComm = collComm->GetLegacy910CollComm();
250 0 : CHK_PTR_NULL(legacyComm);
251 0 : CHK_RET(legacyComm->RegisterOpInfo(opInfo, size));
252 0 : return HCCL_SUCCESS;
253 : }
254 :
255 0 : HcclResult AicpuHcclProcess::AicpuRegOpTaskException(HcommGetOpInfoCallback callback)
256 : {
257 0 : CollCommAicpu* collComm = CollCommAicpuMgr::GetInstance().GetCurrentComm();
258 0 : CHK_PTR_NULL(collComm);
259 0 : HcclCommAicpu* legacyComm = collComm->GetLegacy910CollComm();
260 0 : CHK_PTR_NULL(legacyComm);
261 0 : CHK_RET(legacyComm->RegOpTaskException(callback));
262 0 : return HCCL_SUCCESS;
263 : }
264 :
265 64802 : std::shared_mutex& AicpuHcclProcess::AicpuGetCommMutex() { return CollCommAicpuMgr::GetInstance().GetMutex(); }
266 :
267 30 : hccl::HcclCommAicpu* AicpuHcclProcess::AicpuGetCommbyGroup(const std::string& group)
268 : {
269 30 : CollCommAicpu* collComm = CollCommAicpuMgr::GetInstance().AcquireCommForUse(group);
270 30 : if (collComm == nullptr) {
271 1 : HCCL_ERROR("[AicpuHcclProcess][%s] group[%s] not found", __func__, group.c_str());
272 1 : return nullptr;
273 : }
274 29 : collComm->SetLegacy910CollCommBusy(true);
275 29 : return collComm->GetLegacy910CollComm();
276 : }
277 :
278 0 : bool AicpuHcclProcess::GetCommExecStatus(const std::string& group)
279 : {
280 0 : CollCommAicpu* collComm = CollCommAicpuMgr::GetInstance().FindCommByGroup(group);
281 0 : if (collComm != nullptr) {
282 0 : return collComm->IsLegacy910CollCommBusy();
283 : }
284 0 : return false;
285 : }
286 :
287 28 : void AicpuHcclProcess::AicpuReleaseCommbyGroup(const std::string& group)
288 : {
289 28 : CollCommAicpu* collComm = CollCommAicpuMgr::GetInstance().FindCommByGroup(group);
290 28 : if (collComm != nullptr) {
291 28 : collComm->SetLegacy910CollCommBusy(false);
292 : }
293 28 : CollCommAicpuMgr::GetInstance().ReleaseComm(group);
294 28 : }
295 :
296 5 : u32 AicpuHcclProcess::AicpuRpcClearOpRes(const struct HcclKfcClearOpResTilingData* tilingData)
297 : {
298 5 : if (tilingData == nullptr) {
299 1 : HCCL_ERROR("[AicpuRpcClearOpRes] tilingData is null");
300 1 : return HCCL_E_PARA;
301 : }
302 4 : if (tilingData->magic != HCCL_KFC_CLEAR_OP_RES_MAGIC) {
303 1 : HCCL_ERROR(
304 : "[AicpuRpcClearOpRes] magic mismatch: expect[0x%x] actual[0x%x], reject as misroute",
305 : HCCL_KFC_CLEAR_OP_RES_MAGIC, tilingData->magic);
306 1 : return HCCL_E_PARA;
307 : }
308 3 : if (tilingData->tagCount == 0 || tilingData->tagCount > HCCL_KFC_CLEAR_OP_RES_MAX_BATCH) {
309 2 : HCCL_ERROR(
310 : "[AicpuRpcClearOpRes] invalid tagCount[%u], must be in [1, %u]", tilingData->tagCount,
311 : HCCL_KFC_CLEAR_OP_RES_MAX_BATCH);
312 2 : return HCCL_E_PARA;
313 : }
314 :
315 : // group 来自 host 端共享 HBM；按 C 字符串语义截断，确保不越界
316 1 : char group[HCOMID_MAX_LENGTH + 1] = {0};
317 1 : if (memcpy_s(group, sizeof(group) - 1, tilingData->group, HCOMID_MAX_LENGTH) != EOK) {
318 0 : HCCL_ERROR("[AicpuRpcClearOpRes] memcpy_s group failed");
319 0 : return HCCL_E_MEMORY;
320 : }
321 1 : const std::string groupStr(group);
322 :
323 : // GetCommbyGroup 独占占用槽位（与 ExecOp 互斥），一次覆盖整批 tag 避免循环内反复抢占；超时 10ms 返回 nullptr
324 1 : hccl::HcclCommAicpu* commAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(groupStr);
325 1 : if (commAicpu == nullptr) {
326 1 : HCCL_WARNING(
327 : "[AicpuRpcClearOpRes] group[%s] not found or busy, skip batch; tagCount[%u]", groupStr.c_str(),
328 : tilingData->tagCount);
329 1 : return HCCL_SUCCESS;
330 : }
331 :
332 0 : HcclResult lastErr = HCCL_SUCCESS;
333 0 : for (u32 i = 0; i < tilingData->tagCount; ++i) {
334 0 : char tag[TAG_MAX_LENGTH + 1] = {0};
335 0 : if (memcpy_s(tag, sizeof(tag) - 1, tilingData->tags[i], TAG_MAX_LENGTH) != EOK) {
336 0 : HCCL_ERROR("[AicpuRpcClearOpRes] memcpy_s tag failed, group[%s] idx[%u]", groupStr.c_str(), i);
337 0 : lastErr = HCCL_E_MEMORY;
338 0 : continue;
339 : }
340 0 : const std::string tagStr(tag);
341 0 : HcclResult ret = commAicpu->ClearOpResource(tagStr);
342 0 : if (ret != HCCL_SUCCESS) {
343 0 : HCCL_ERROR(
344 : "[AicpuRpcClearOpRes] ClearOpResource fail, group[%s] tag[%s] ret[%d]", groupStr.c_str(),
345 : tagStr.c_str(), ret);
346 0 : lastErr = ret;
347 : // 单 tag 失败不影响后续 tag, 尽量清干净
348 : }
349 0 : }
350 0 : AicpuHcclProcess::AicpuReleaseCommbyGroup(groupStr);
351 0 : HCCL_INFO(
352 : "[AicpuRpcClearOpRes] group[%s] processed batch tagCount[%u] lastErr[%d]", groupStr.c_str(),
353 : tilingData->tagCount, lastErr);
354 0 : return static_cast<u32>(lastErr);
355 1 : }
356 :
357 64802 : HcclResult AicpuHcclProcess::AicpuGetCommAll(std::vector<std::pair<std::string, HcclCommAicpu*>>& aicpuCommInfo)
358 : {
359 64802 : std::vector<std::pair<std::string, CollCommAicpu*>> commInfo;
360 64802 : CollCommAicpuMgr::GetInstance().GetAllComms(commInfo);
361 64802 : for (auto& kv : commInfo) {
362 0 : HcclCommAicpu* legacy = kv.second->GetLegacy910CollComm();
363 0 : if (legacy != nullptr) {
364 0 : aicpuCommInfo.push_back({kv.first, legacy});
365 : }
366 : }
367 64802 : return HCCL_SUCCESS;
368 64802 : }
369 :
370 0 : void AicpuHcclProcess::AicpuDestoryCommbyGroup(const std::string& group)
371 : {
372 0 : CollCommAicpuMgr::GetInstance().DestroyComm(group);
373 0 : }
374 :
375 0 : HcclResult AicpuHcclProcess::HandleOneSideService(const OpTilingData* tilingData)
376 : {
377 0 : return HcclOneSideServiceAicpu::Process(tilingData);
378 : }
379 :
380 0 : HcclResult AicpuHcclProcess::AicpuRunRpcServerV2(
381 : hccl::HcclCommAicpu* hcclCommAicpu, OpTilingData* tilingData, HcclOpResParam* commParam)
382 : {
383 0 : std::string algName = tilingData->algName;
384 0 : std::string tag = reinterpret_cast<char*>(tilingData->tag);
385 0 : std::string newTag = reinterpret_cast<char*>(tilingData->newTag);
386 :
387 0 : HCCL_DEBUG(
388 : "[AicpuHcclProcess][AicpuRunRpcServerV2]Entry AicpuRunRpcServerV2, group[%s], tag[%s], newTag[%s]",
389 : hcclCommAicpu->GetGroupName().c_str(), tag.c_str(), newTag.c_str());
390 :
391 0 : HCCL_DEBUG(
392 : "[AicpuHcclProcess][AicpuRunRpcServerV2]Entry AicpuRunRpcServerV2, algName[%s], algtype[%llu],"
393 : "floatOverflowMode[%u], dumpDebug[%u], debugMode[%u], inputPtr[%p], outputPtr[%p].",
394 : algName.c_str(), tilingData->algType, tilingData->floatOverflowMode, tilingData->dumpDebug,
395 : tilingData->debugMode, tilingData->inputPtr, tilingData->outputPtr);
396 :
397 0 : HCCL_DEBUG(
398 : "[AicpuHcclProcess][AicpuRunRpcServerV2]Entry AicpuRunRpcServerV2, reduceType[%u], syncMode[%u],"
399 : "root[%u], dstRank[%u], srcRank[%u], opType[%u], index[%u], length[%llu].",
400 : tilingData->reduceType, tilingData->syncMode, tilingData->root, tilingData->dstRank, tilingData->srcRank,
401 : tilingData->opType, tilingData->index, tilingData->length);
402 :
403 0 : HCCL_DEBUG(
404 : "[AicpuHcclProcess][AicpuRunRpcServerV2]Entry AicpuRunRpcServerV2, aicpuCacheEnable[%u]",
405 : tilingData->aicpuCacheEnable);
406 :
407 0 : CHK_RET(hcclCommAicpu->RecordHostOrder(commParam, tag, tilingData->orderLaunchMode));
408 :
409 0 : hccl::OpParam opParam;
410 0 : opParam.tag = tag;
411 0 : opParam.inputPtr = reinterpret_cast<void*>(tilingData->inputPtr);
412 0 : opParam.outputPtr = reinterpret_cast<void*>(tilingData->outputPtr);
413 0 : opParam.reduceType = static_cast<HcclReduceOp>(tilingData->reduceType);
414 0 : opParam.stream = hcclCommAicpu->GetMainStream();
415 0 : opParam.syncMode = static_cast<SyncMode>(tilingData->syncMode);
416 0 : opParam.inputSymWindow = reinterpret_cast<void*>(tilingData->inputSymWindow);
417 0 : opParam.inputOffset = tilingData->inputOffset;
418 0 : opParam.outputSymWindow = reinterpret_cast<void*>(tilingData->outputSymWindow);
419 0 : opParam.outputOffset = tilingData->outputOffset;
420 :
421 0 : hcclCommAicpu->UpdateNotifyWaitTimeOut(opParam.syncMode, commParam->config.notifyWaitTime);
422 :
423 0 : opParam.opBaseAtraceInfo = nullptr;
424 0 : opParam.root = tilingData->root;
425 0 : opParam.dstRank = tilingData->dstRank;
426 0 : opParam.srcRank = tilingData->srcRank;
427 0 : opParam.opType = static_cast<HcclCMDType>(tilingData->opType);
428 0 : opParam.isZeroCopy = tilingData->isZeroCopy;
429 0 : opParam.supportSymmetricMemory = tilingData->isSymmetricMemory;
430 0 : opParam.index = tilingData->index;
431 0 : opParam.isCapture = tilingData->isCapture;
432 0 : opParam.needIncreLink = tilingData->needIncreLink;
433 0 : opParam.aicpuCacheEnable = tilingData->aicpuCacheEnable;
434 0 : hcclCommAicpu->PrepareOpRetryHandler(
435 0 : tilingData->inplaceSupportRetry, tilingData->retryEnable, tilingData->inPlaceSupportRetryStatus,
436 0 : tilingData->isInplacePreSync, tilingData->isPostSync);
437 0 : u8* dynamicDataPtr = reinterpret_cast<u8*>(tilingData) + sizeof(struct OpTilingData);
438 : char stackLogBuffer[LOG_TMPBUF_SIZE];
439 0 : if (opParam.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV) {
440 0 : struct OpTilingBatchSendRecvDataDes* batchSendRecvDataPtr
441 : = reinterpret_cast<struct OpTilingBatchSendRecvDataDes*>(dynamicDataPtr);
442 0 : opParam.BatchSendRecvDataDes.itemNum = batchSendRecvDataPtr->itemNum;
443 0 : opParam.BatchSendRecvDataDes.sendRecvItemsPtr = batchSendRecvDataPtr->batchSendRecvItem;
444 : opParam.BatchSendRecvDataDes.isDirectRemoteRank
445 0 : = reinterpret_cast<u8*>(batchSendRecvDataPtr->batchSendRecvItem + batchSendRecvDataPtr->itemNum);
446 0 : s32 ret = snprintf_s(stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s", opParam.tag.c_str());
447 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
448 0 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALL) {
449 0 : struct OpTilingAllToAllDataDes* a2ADataPtr = reinterpret_cast<struct OpTilingAllToAllDataDes*>(dynamicDataPtr);
450 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(a2ADataPtr->sendType);
451 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(a2ADataPtr->recvType);
452 0 : opParam.All2AllDataDes.sendCount = a2ADataPtr->sendCount;
453 0 : HCCL_DEBUG(
454 : "[AicpuHcclProcess][AicpuRunRpcServerV2] AllToAll aicpu, sendCounts[%llu] rankSize[%u]",
455 : opParam.All2AllDataDes.sendCount, commParam->rankSize);
456 0 : s32 ret = snprintf_s(
457 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
458 0 : opParam.All2AllDataDes.sendCount, opParam.All2AllDataDes.sendType);
459 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
460 0 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) {
461 0 : struct OpTilingAlltoallvDataDes* alltoallvDataPtr
462 : = reinterpret_cast<struct OpTilingAlltoallvDataDes*>(dynamicDataPtr);
463 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(alltoallvDataPtr->sendType);
464 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(alltoallvDataPtr->recvType);
465 0 : u64 rankSize = commParam->rankSize;
466 0 : opParam.All2AllDataDes.sendCounts = static_cast<void*>(alltoallvDataPtr->sendRecvInfos);
467 : opParam.All2AllDataDes.recvCounts
468 0 : = static_cast<void*>(static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + rankSize);
469 0 : opParam.All2AllDataDes.sdispls = static_cast<void*>(
470 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_2 * rankSize);
471 0 : opParam.All2AllDataDes.rdispls = static_cast<void*>(
472 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_3 * rankSize);
473 0 : if (algName == "RunAlltoAllVTwoLevelPipeline") {
474 0 : hcclCommAicpu->SetSendRecvInfoPtr(static_cast<void*>(
475 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_4 * rankSize));
476 : }
477 0 : HCCL_DEBUG(
478 : "[AicpuHcclProcess][AicpuRunRpcServerV2] sendCountsPtr[%p], recvCountsPtr[%p], sdisplsPtr[%p], "
479 : "rdisplsPtr[%p].",
480 : opParam.All2AllDataDes.sendCounts, opParam.All2AllDataDes.recvCounts, opParam.All2AllDataDes.sdispls,
481 : opParam.All2AllDataDes.rdispls);
482 0 : for (u32 i = 0; i < rankSize; i++) {
483 0 : HCCL_DEBUG(
484 : "[AicpuHcclProcess][AicpuRunRpcServerV2] sendCounts[%llu], recvCounts[%llu].",
485 : *(static_cast<const u64*>(opParam.All2AllDataDes.sendCounts) + i),
486 : *(static_cast<const u64*>(opParam.All2AllDataDes.recvCounts) + i));
487 : }
488 0 : s32 ret = snprintf_s(
489 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
490 0 : opParam.All2AllDataDes.sendCounts, opParam.All2AllDataDes.sendType);
491 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
492 0 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) {
493 0 : struct OpTilingAlltoallvcDataDes* alltoallvcDataPtr
494 : = reinterpret_cast<struct OpTilingAlltoallvcDataDes*>(dynamicDataPtr);
495 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(alltoallvcDataPtr->sendType);
496 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(alltoallvcDataPtr->recvType);
497 0 : opParam.All2AllDataDes.sendCountMatrix = static_cast<void*>(alltoallvcDataPtr->sendCountMatrix);
498 0 : s32 ret = snprintf_s(
499 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
500 0 : opParam.All2AllDataDes.sendCountMatrix, opParam.All2AllDataDes.sendType);
501 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
502 0 : } else if (
503 0 : opParam.opType == HcclCMDType::HCCL_CMD_ALLGATHER_V
504 0 : || opParam.opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V) {
505 0 : const u64 vStructSize = sizeof(struct OpTilingVDataDes);
506 0 : CHK_PRT_RET(
507 : tilingData->length < vStructSize,
508 : HCCL_ERROR(
509 : "[AicpuHcclProcess][AicpuRunRpcServerV2] TilingData "
510 : "corrupt, length[%llu], expect[%llu]",
511 : tilingData->length, vStructSize),
512 : HCCL_E_PARA);
513 0 : const u32 rankSize = commParam->rankSize;
514 0 : struct OpTilingVDataDes* vDataPtr = reinterpret_cast<struct OpTilingVDataDes*>(dynamicDataPtr);
515 0 : const u64 vDataLen = CalcOpTilingVDataDesVDataLen(rankSize);
516 0 : CHK_PRT_RET(
517 : vDataPtr->vDataLen != vDataLen,
518 : HCCL_ERROR(
519 : "[AicpuHcclProcess][AicpuRunRpcServerV2] TilingVDataDes "
520 : "corrupt, length[%llu], expect[%llu], rankSize[%u]",
521 : vDataPtr->vDataLen, vDataLen, rankSize),
522 : HCCL_E_PARA);
523 0 : opParam.VDataDes.dataType = static_cast<HcclDataType>(vDataPtr->dataType);
524 0 : opParam.VDataDes.counts = static_cast<void*>(static_cast<u64*>(vDataPtr->vData));
525 0 : opParam.VDataDes.displs = static_cast<void*>(static_cast<u64*>(vDataPtr->vData) + rankSize);
526 0 : CHK_RET(CalcDataSizeV(opParam, rankSize));
527 0 : s32 ret = snprintf_s(
528 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,ds:%llu,dt:%u", opParam.tag.c_str(),
529 0 : opParam.VDataDes.counts, opParam.VDataDes.displs, opParam.VDataDes.dataType);
530 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
531 0 : } else {
532 0 : struct OpTilingDataDes* opDataDesPtr = reinterpret_cast<struct OpTilingDataDes*>(dynamicDataPtr);
533 0 : opParam.DataDes.count = opDataDesPtr->count;
534 0 : opParam.DataDes.dataType = static_cast<HcclDataType>(opDataDesPtr->dataType);
535 0 : CHK_RET(CalcDataSize(
536 : opParam.opType, static_cast<HcclDataType>(opDataDesPtr->dataType), opDataDesPtr->count,
537 : hcclCommAicpu->GetRankSize(), opParam.inputSize, opParam.outputSize));
538 0 : HCCL_DEBUG(
539 : "[AicpuHcclProcess][AicpuRunRpcServerV2] Entry AicpuRunRpcServerV2, "
540 : "count[%llu], dataType[%u] inputSize[%lu] outputSize[%lu].",
541 : opDataDesPtr->count, opDataDesPtr->dataType, opParam.inputSize, opParam.outputSize);
542 0 : s32 ret = snprintf_s(
543 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
544 0 : opParam.DataDes.count, opParam.DataDes.dataType);
545 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
546 : }
547 :
548 0 : CHK_RET(hrtSetLocalDeviceSatMode(static_cast<aclrtFloatOverflowMode>(tilingData->floatOverflowMode)));
549 0 : hcclCommAicpu->SetDumpDebug(tilingData->dumpDebug);
550 0 : hcclCommAicpu->SetAlgType(tilingData->algType);
551 0 : hcclCommAicpu->SetDebugMode(tilingData->debugMode);
552 0 : hcclCommAicpu->SetIsDeviceMode(false);
553 0 : hcclCommAicpu->SetUserStreamId(tilingData->userStreamId);
554 0 : CHK_RET(hcclCommAicpu->ParseHierarchicalAlgOption(tilingData->ahcConfInfo));
555 : /* 接口交互信息日志 */
556 0 : std::string logInfo = std::string(stackLogBuffer);
557 0 : CHK_RET_AND_PRINT_IDE(hcclCommAicpu->SaveTraceInfo(logInfo), opParam.tag.c_str());
558 :
559 0 : HcclUs startut = TIME_NOW();
560 0 : HcclResult ret = hcclCommAicpu->ExecOp(newTag, algName, opParam, commParam);
561 0 : CHK_PRT_RET(
562 : ret != HCCL_SUCCESS,
563 : HCCL_ERROR("[AicpuHcclProcess][AicpuRunRpcServerV2] newTag[%s] algName[%s]", newTag.c_str(), algName.c_str()),
564 : ret);
565 0 : HcclUs endut = TIME_NOW();
566 : /* 关键状态记录 */
567 : std::string endInfo
568 0 : = "AicpuRunRpcServerV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us";
569 0 : CHK_RET_AND_PRINT_IDE(hcclCommAicpu->SaveTraceInfo(endInfo), opParam.tag.c_str());
570 :
571 0 : HCCL_INFO("[AicpuHcclProcess][AicpuRunRpcServerV2]AicpuRunRpcServerV2 process end-------");
572 0 : return HCCL_SUCCESS;
573 0 : }
574 :
575 80 : void AicpuHcclProcess::CopyCtxInfo(AicpuComContext* ctx)
576 : {
577 80 : auto otherClusterId = CLUSTER_CNT - ctx->clusterId - 1;
578 80 : auto otherCluster = &g_comContext[otherClusterId];
579 80 : *otherCluster = *ctx;
580 80 : otherCluster->clusterId = otherClusterId;
581 80 : HCCL_DEBUG("curClusterId = %d, otherClusterId = %d, copy finished", ctx->clusterId, otherCluster->clusterId);
582 80 : }
583 :
584 1 : void AicpuHcclProcess::CopyCtxForBackGroundDfx(const AicpuComContext* ctx)
585 : {
586 1 : auto otherClusterId = CLUSTER_CNT - ctx->clusterId - 1;
587 1 : AicpuComContext* otherCluster = &g_comContext[otherClusterId];
588 1 : otherCluster->workSpaceAddr = ctx->workSpaceAddr;
589 1 : otherCluster->notifyOff = ctx->notifyOff;
590 1 : otherCluster->notifyBeginCnt = ctx->notifyBeginCnt;
591 1 : otherCluster->totalTurnCnt = ctx->totalTurnCnt;
592 1 : AicpuSqeContext::SaveVariable();
593 1 : }
594 :
595 1 : HcclResult AicpuHcclProcess::WaitAsyncFlag(
596 : hccl::Transport::Buffer* localFlagBufforCheck, const uint32_t flagValue, uint64_t timeOut)
597 : {
598 : // 轮询等待flag
599 1 : if (flagValue < FLAG_OFFSET) {
600 0 : HCCL_ERROR("[AicpuHcclProcess][WaitAsyncFlag] flagValue must > 0,now flagValue is [%u]", flagValue);
601 0 : return HCCL_E_PARA;
602 : }
603 1 : uint32_t index = flagValue - FLAG_OFFSET;
604 1 : bool isTimeout = true;
605 1 : u64 startTime = GetCurCpuTimestamp();
606 1 : CHK_PTR_NULL(localFlagBufforCheck);
607 0 : CHK_PTR_NULL(localFlagBufforCheck[index].addr);
608 0 : uint32_t* waitPtr = const_cast<uint32_t*>(static_cast<const uint32_t*>(localFlagBufforCheck[index].addr));
609 0 : while ((GetCurCpuTimestamp() - startTime) < static_cast<unsigned long long>(NSEC_PER_SEC * timeOut)) {
610 0 : if (*waitPtr == flagValue) {
611 0 : isTimeout = false;
612 0 : *waitPtr = 0;
613 0 : break; // 当前位置的值等于flag时，说明对端写过来了，可以退出
614 : }
615 : }
616 0 : CHK_PRT_RET(
617 : isTimeout,
618 : HCCL_ERROR(
619 : "[AicpuHcclProcess][WaitAsyncFlag]Kernel Run TimeOut %llus, now Opetation is CheckFlag, "
620 : "localFlagBufforCheck0.addr is [%p], localFlagBufforCheck0.value is [%u],"
621 : "localFlagBufforCheck1.addr is [%p], localFlagBufforCheck1.value is [%u],"
622 : "localFlagBufforCheck2.addr is [%p], localFlagBufforCheck2.value is [%u]",
623 : timeOut, localFlagBufforCheck[0].addr, *(const uint32_t*)(localFlagBufforCheck[0].addr),
624 : localFlagBufforCheck[1].addr, *(const uint32_t*)(localFlagBufforCheck[1].addr),
625 : localFlagBufforCheck[2].addr, *(const uint32_t*)(localFlagBufforCheck[2].addr)),
626 : HCCL_E_INTERNAL);
627 0 : return HCCL_SUCCESS;
628 : }
629 :
630 0 : HcclResult AicpuHcclProcess::InitAsyncFlag(
631 : const uint32_t* lFlagAddr, const uint32_t* rFlagAddr, hccl::Transport::Buffer* localFlagBufforCheck,
632 : hccl::Transport::Buffer* localFlagBufforWrite, hccl::Transport::Buffer* remoteFlagBuf)
633 : {
634 0 : CHK_PTR_NULL(lFlagAddr);
635 0 : CHK_PTR_NULL(rFlagAddr);
636 0 : CHK_PTR_NULL(remoteFlagBuf);
637 0 : for (uint32_t i = 0; i < POST_SEND_FLAG_COUNT; i++) {
638 0 : localFlagBufforCheck[i].addr = lFlagAddr + i * FLAG_INTERVAL; // 接收对端给本端写的flag
639 0 : localFlagBufforCheck[i].size = sizeof(uint32_t);
640 0 : localFlagBufforWrite[i].addr = lFlagAddr + i * FLAG_INTERVAL + FLAG_OFFSET; // 存放本端给对端写的flag
641 0 : localFlagBufforWrite[i].size = sizeof(uint32_t);
642 0 : remoteFlagBuf[i].addr = rFlagAddr + i * FLAG_INTERVAL;
643 0 : remoteFlagBuf[i].size = sizeof(uint32_t);
644 0 : uint32_t* waitPtr = const_cast<uint32_t*>(static_cast<const uint32_t*>(localFlagBufforWrite[i].addr));
645 0 : *waitPtr = i + FLAG_OFFSET; // 置flag
646 : }
647 0 : return HCCL_SUCCESS;
648 : }
649 :
650 0 : HcclResult AicpuHcclProcess::AicpuIndOpThreadInit(ThreadMgrAicpuParam* param)
651 : {
652 0 : std::string group = param->hcomId;
653 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
654 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
655 0 : HcclResult ret = hcclCommAicpu->InitThreads(param);
656 0 : CHK_PRT_RET(
657 : ret != HCCL_SUCCESS,
658 : HCCL_ERROR(
659 : "[AicpuHcclProcess][AicpuIndOpThreadInit]errNo[0x%016llx] Failed to init threads group[%s]",
660 : HCCL_ERROR_CODE(ret), group.c_str()),
661 : ret);
662 0 : AicpuReleaseCommbyGroup(group);
663 0 : return HCCL_SUCCESS;
664 0 : }
665 :
666 0 : HcclResult AicpuHcclProcess::AicpuIndOpChannelInit(HcclIndOpChannelRemoteResV3* commParam)
667 : {
668 0 : std::string group = commParam->hcomId;
669 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
670 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
671 0 : HcclResult ret = hcclCommAicpu->AllocChannelResource(commParam);
672 0 : CHK_PRT_RET(
673 : ret != HCCL_SUCCESS,
674 : HCCL_ERROR(
675 : "[AicpuHcclProcess][AicpuIndOpChannelInit]errNo[0x%016llx] Failed to init channels group[%s]",
676 : HCCL_ERROR_CODE(ret), group.c_str()),
677 : ret);
678 0 : AicpuReleaseCommbyGroup(group);
679 0 : return HCCL_SUCCESS;
680 0 : }
681 :
682 1 : HcclResult AicpuHcclProcess::AicpuIndOpNotifyInit(NotifyMgrAicpuParam* param)
683 : {
684 1 : CHK_PTR_NULL(param);
685 0 : std::string group = param->hcomId;
686 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
687 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
688 :
689 0 : HcclResult ret = HCCL_E_INTERNAL;
690 0 : if (param->freeFlag) {
691 0 : ret = hcclCommAicpu->NotifyFree(param);
692 0 : CHK_PRT_RET(
693 : ret != HCCL_SUCCESS,
694 : HCCL_ERROR(
695 : "[AicpuHcclProcess][%s]errNo[0x%016llx] Failed to free notifys group[%s]", __func__,
696 : HCCL_ERROR_CODE(ret), group.c_str()),
697 : ret);
698 : } else {
699 0 : ret = hcclCommAicpu->NotifyAlloc(param);
700 0 : CHK_PRT_RET(
701 : ret != HCCL_SUCCESS,
702 : HCCL_ERROR(
703 : "[AicpuHcclProcess][%s]errNo[0x%016llx] Failed to alloc notifys group[%s]", __func__,
704 : HCCL_ERROR_CODE(ret), group.c_str()),
705 : ret);
706 : }
707 :
708 0 : HCCL_INFO(
709 : "[AicpuHcclProcess][%s] comm identifier[%s], notify op[%u] success, num[%u]", __func__, group.c_str(),
710 : param->freeFlag, param->notifyNum);
711 0 : AicpuReleaseCommbyGroup(group);
712 0 : return HCCL_SUCCESS;
713 0 : }
|
