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 226360872 : void AicpuGetAllComContext(AicpuComContext*& contextBase, uint32_t& contextNum)
48 : {
49 226360872 : contextBase = &g_comContext[0];
50 226360872 : contextNum = CLUSTER_CNT;
51 226360872 : 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 6 : HcclResult AicpuHcclProcess::CalcDataSize(
72 : HcclCMDType op, HcclDataType type, u64 count, u32 rankSize, u64& inputSize, u64& outputSize)
73 : {
74 6 : u32 perDataSize = DataUnitSize(type);
75 6 : if (perDataSize == 0) {
76 1 : HCCL_ERROR("[AicpuHcclProcess][CalcDataSize] type [%u] DataUnitSize is 0", type);
77 1 : return HCCL_E_PARA;
78 : }
79 :
80 5 : switch (op) {
81 3 : case HcclCMDType::HCCL_CMD_ALLGATHER:
82 3 : inputSize = count * perDataSize;
83 3 : outputSize = rankSize * count * perDataSize;
84 3 : 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 5 : HCCL_DEBUG(
102 : "[AicpuHcclProcess][CalcDataSize] perDataSize %u count %lu rankSize %u input %lu output %lu", perDataSize,
103 : count, rankSize, inputSize, outputSize);
104 5 : 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 2 : HcclResult AicpuHcclProcess::AicpuRunRpcServerV2(
381 : hccl::HcclCommAicpu* hcclCommAicpu, OpTilingData* tilingData, HcclOpResParam* commParam)
382 : {
383 4 : std::string algName = tilingData->algName;
384 4 : std::string tag = reinterpret_cast<char*>(tilingData->tag);
385 2 : std::string newTag = reinterpret_cast<char*>(tilingData->newTag);
386 :
387 2 : 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 2 : 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 2 : 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 2 : HCCL_DEBUG(
404 : "[AicpuHcclProcess][AicpuRunRpcServerV2]Entry AicpuRunRpcServerV2, aicpuCacheEnable[%u]",
405 : tilingData->aicpuCacheEnable);
406 :
407 2 : CHK_RET(hcclCommAicpu->RecordHostOrder(commParam, tag, tilingData->orderLaunchMode));
408 :
409 2 : hccl::OpParam opParam;
410 2 : opParam.tag = tag;
411 2 : opParam.inputPtr = reinterpret_cast<void*>(tilingData->inputPtr);
412 2 : opParam.outputPtr = reinterpret_cast<void*>(tilingData->outputPtr);
413 2 : opParam.reduceType = static_cast<HcclReduceOp>(tilingData->reduceType);
414 2 : opParam.stream = hcclCommAicpu->GetMainStream();
415 2 : opParam.syncMode = static_cast<SyncMode>(tilingData->syncMode);
416 2 : opParam.inputSymWindow = reinterpret_cast<void*>(tilingData->inputSymWindow);
417 2 : opParam.inputOffset = tilingData->inputOffset;
418 2 : opParam.outputSymWindow = reinterpret_cast<void*>(tilingData->outputSymWindow);
419 2 : opParam.outputOffset = tilingData->outputOffset;
420 :
421 2 : hcclCommAicpu->UpdateNotifyWaitTimeOut(opParam.syncMode, commParam->config.notifyWaitTime);
422 :
423 2 : opParam.opBaseAtraceInfo = nullptr;
424 2 : opParam.root = tilingData->root;
425 2 : opParam.dstRank = tilingData->dstRank;
426 2 : opParam.srcRank = tilingData->srcRank;
427 2 : opParam.opType = static_cast<HcclCMDType>(tilingData->opType);
428 2 : opParam.isZeroCopy = tilingData->isZeroCopy;
429 2 : opParam.supportSymmetricMemory = tilingData->isSymmetricMemory;
430 2 : opParam.index = tilingData->index;
431 2 : opParam.isCapture = tilingData->isCapture;
432 2 : opParam.needIncreLink = tilingData->needIncreLink;
433 2 : opParam.aicpuCacheEnable = tilingData->aicpuCacheEnable;
434 2 : opParam.aicpuUnfoldMode = tilingData->aicpuUnfoldMode;
435 2 : hcclCommAicpu->PrepareOpRetryHandler(
436 2 : tilingData->inplaceSupportRetry, tilingData->retryEnable, tilingData->inPlaceSupportRetryStatus,
437 2 : tilingData->isInplacePreSync, tilingData->isPostSync);
438 2 : u8* dynamicDataPtr = reinterpret_cast<u8*>(tilingData) + sizeof(struct OpTilingData);
439 : char stackLogBuffer[LOG_TMPBUF_SIZE];
440 2 : if (opParam.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV) {
441 0 : struct OpTilingBatchSendRecvDataDes* batchSendRecvDataPtr
442 : = reinterpret_cast<struct OpTilingBatchSendRecvDataDes*>(dynamicDataPtr);
443 0 : opParam.BatchSendRecvDataDes.itemNum = batchSendRecvDataPtr->itemNum;
444 0 : opParam.BatchSendRecvDataDes.sendRecvItemsPtr = batchSendRecvDataPtr->batchSendRecvItem;
445 : opParam.BatchSendRecvDataDes.isDirectRemoteRank
446 0 : = reinterpret_cast<u8*>(batchSendRecvDataPtr->batchSendRecvItem + batchSendRecvDataPtr->itemNum);
447 0 : s32 ret = snprintf_s(stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s", opParam.tag.c_str());
448 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
449 2 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALL) {
450 0 : struct OpTilingAllToAllDataDes* a2ADataPtr = reinterpret_cast<struct OpTilingAllToAllDataDes*>(dynamicDataPtr);
451 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(a2ADataPtr->sendType);
452 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(a2ADataPtr->recvType);
453 0 : opParam.All2AllDataDes.sendCount = a2ADataPtr->sendCount;
454 0 : HCCL_DEBUG(
455 : "[AicpuHcclProcess][AicpuRunRpcServerV2] AllToAll aicpu, sendCounts[%llu] rankSize[%u]",
456 : opParam.All2AllDataDes.sendCount, commParam->rankSize);
457 0 : s32 ret = snprintf_s(
458 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
459 0 : opParam.All2AllDataDes.sendCount, opParam.All2AllDataDes.sendType);
460 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
461 2 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) {
462 0 : struct OpTilingAlltoallvDataDes* alltoallvDataPtr
463 : = reinterpret_cast<struct OpTilingAlltoallvDataDes*>(dynamicDataPtr);
464 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(alltoallvDataPtr->sendType);
465 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(alltoallvDataPtr->recvType);
466 0 : u64 rankSize = commParam->rankSize;
467 0 : opParam.All2AllDataDes.sendCounts = static_cast<void*>(alltoallvDataPtr->sendRecvInfos);
468 : opParam.All2AllDataDes.recvCounts
469 0 : = static_cast<void*>(static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + rankSize);
470 0 : opParam.All2AllDataDes.sdispls = static_cast<void*>(
471 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_2 * rankSize);
472 0 : opParam.All2AllDataDes.rdispls = static_cast<void*>(
473 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_3 * rankSize);
474 0 : if (algName == "RunAlltoAllVTwoLevelPipeline") {
475 0 : hcclCommAicpu->SetSendRecvInfoPtr(static_cast<void*>(
476 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_INFO_INDEX_4 * rankSize));
477 : }
478 0 : HCCL_DEBUG(
479 : "[AicpuHcclProcess][AicpuRunRpcServerV2] sendCountsPtr[%p], recvCountsPtr[%p], sdisplsPtr[%p], "
480 : "rdisplsPtr[%p].",
481 : opParam.All2AllDataDes.sendCounts, opParam.All2AllDataDes.recvCounts, opParam.All2AllDataDes.sdispls,
482 : opParam.All2AllDataDes.rdispls);
483 0 : for (u32 i = 0; i < rankSize; i++) {
484 0 : HCCL_DEBUG(
485 : "[AicpuHcclProcess][AicpuRunRpcServerV2] sendCounts[%llu], recvCounts[%llu].",
486 : *(static_cast<const u64*>(opParam.All2AllDataDes.sendCounts) + i),
487 : *(static_cast<const u64*>(opParam.All2AllDataDes.recvCounts) + i));
488 : }
489 0 : s32 ret = snprintf_s(
490 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
491 0 : opParam.All2AllDataDes.sendCounts, opParam.All2AllDataDes.sendType);
492 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
493 2 : } else if (opParam.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) {
494 0 : struct OpTilingAlltoallvcDataDes* alltoallvcDataPtr
495 : = reinterpret_cast<struct OpTilingAlltoallvcDataDes*>(dynamicDataPtr);
496 0 : opParam.All2AllDataDes.sendType = static_cast<HcclDataType>(alltoallvcDataPtr->sendType);
497 0 : opParam.All2AllDataDes.recvType = static_cast<HcclDataType>(alltoallvcDataPtr->recvType);
498 0 : opParam.All2AllDataDes.sendCountMatrix = static_cast<void*>(alltoallvcDataPtr->sendCountMatrix);
499 0 : s32 ret = snprintf_s(
500 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
501 0 : opParam.All2AllDataDes.sendCountMatrix, opParam.All2AllDataDes.sendType);
502 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
503 2 : } else if (
504 2 : opParam.opType == HcclCMDType::HCCL_CMD_ALLGATHER_V
505 2 : || opParam.opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V) {
506 0 : const u64 vStructSize = sizeof(struct OpTilingVDataDes);
507 0 : CHK_PRT_RET(
508 : tilingData->length < vStructSize,
509 : HCCL_ERROR(
510 : "[AicpuHcclProcess][AicpuRunRpcServerV2] TilingData "
511 : "corrupt, length[%llu], expect[%llu]",
512 : tilingData->length, vStructSize),
513 : HCCL_E_PARA);
514 0 : const u32 rankSize = commParam->rankSize;
515 0 : struct OpTilingVDataDes* vDataPtr = reinterpret_cast<struct OpTilingVDataDes*>(dynamicDataPtr);
516 0 : const u64 vDataLen = CalcOpTilingVDataDesVDataLen(rankSize);
517 0 : CHK_PRT_RET(
518 : vDataPtr->vDataLen != vDataLen,
519 : HCCL_ERROR(
520 : "[AicpuHcclProcess][AicpuRunRpcServerV2] TilingVDataDes "
521 : "corrupt, length[%llu], expect[%llu], rankSize[%u]",
522 : vDataPtr->vDataLen, vDataLen, rankSize),
523 : HCCL_E_PARA);
524 0 : opParam.VDataDes.dataType = static_cast<HcclDataType>(vDataPtr->dataType);
525 0 : opParam.VDataDes.counts = static_cast<void*>(static_cast<u64*>(vDataPtr->vData));
526 0 : opParam.VDataDes.displs = static_cast<void*>(static_cast<u64*>(vDataPtr->vData) + rankSize);
527 0 : CHK_RET(CalcDataSizeV(opParam, rankSize));
528 0 : s32 ret = snprintf_s(
529 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,ds:%llu,dt:%u", opParam.tag.c_str(),
530 0 : opParam.VDataDes.counts, opParam.VDataDes.displs, opParam.VDataDes.dataType);
531 0 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
532 0 : } else {
533 2 : struct OpTilingDataDes* opDataDesPtr = reinterpret_cast<struct OpTilingDataDes*>(dynamicDataPtr);
534 2 : opParam.DataDes.count = opDataDesPtr->count;
535 2 : opParam.DataDes.dataType = static_cast<HcclDataType>(opDataDesPtr->dataType);
536 2 : CHK_RET(CalcDataSize(
537 : opParam.opType, static_cast<HcclDataType>(opDataDesPtr->dataType), opDataDesPtr->count,
538 : hcclCommAicpu->GetRankSize(), opParam.inputSize, opParam.outputSize));
539 2 : HCCL_DEBUG(
540 : "[AicpuHcclProcess][AicpuRunRpcServerV2] Entry AicpuRunRpcServerV2, "
541 : "count[%llu], dataType[%u] inputSize[%lu] outputSize[%lu].",
542 : opDataDesPtr->count, opDataDesPtr->dataType, opParam.inputSize, opParam.outputSize);
543 2 : s32 ret = snprintf_s(
544 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U, "tag:%s,ct:%llu,dt:%u", opParam.tag.c_str(),
545 2 : opParam.DataDes.count, opParam.DataDes.dataType);
546 2 : CHK_PRT_CONT(ret == -1, HCCL_WARNING("Failed to build log info, tag[%s].", opParam.tag.c_str()));
547 : }
548 :
549 2 : CHK_RET(hrtSetLocalDeviceSatMode(static_cast<aclrtFloatOverflowMode>(tilingData->floatOverflowMode)));
550 2 : hcclCommAicpu->SetDumpDebug(tilingData->dumpDebug);
551 2 : hcclCommAicpu->SetAlgType(tilingData->algType);
552 2 : hcclCommAicpu->SetDebugMode(tilingData->debugMode);
553 2 : hcclCommAicpu->SetIsDeviceMode(false);
554 2 : hcclCommAicpu->SetUserStreamId(tilingData->userStreamId);
555 2 : CHK_RET(hcclCommAicpu->ParseHierarchicalAlgOption(tilingData->ahcConfInfo));
556 : /* 接口交互信息日志 */
557 2 : std::string logInfo = std::string(stackLogBuffer);
558 2 : CHK_RET_AND_PRINT_IDE(hcclCommAicpu->SaveTraceInfo(logInfo), opParam.tag.c_str());
559 :
560 2 : HcclUs startut = TIME_NOW();
561 2 : HcclResult ret = hcclCommAicpu->ExecOp(newTag, algName, opParam, commParam);
562 2 : CHK_PRT_RET(
563 : ret != HCCL_SUCCESS,
564 : HCCL_ERROR("[AicpuHcclProcess][AicpuRunRpcServerV2] newTag[%s] algName[%s]", newTag.c_str(), algName.c_str()),
565 : ret);
566 2 : HcclUs endut = TIME_NOW();
567 : /* 关键状态记录 */
568 : std::string endInfo
569 2 : = "AicpuRunRpcServerV2:success,take time: " + std::to_string(DURATION_US(endut - startut).count()) + " us";
570 2 : CHK_RET_AND_PRINT_IDE(hcclCommAicpu->SaveTraceInfo(endInfo), opParam.tag.c_str());
571 :
572 2 : HCCL_INFO("[AicpuHcclProcess][AicpuRunRpcServerV2]AicpuRunRpcServerV2 process end-------");
573 2 : return HCCL_SUCCESS;
574 2 : }
575 :
576 80 : void AicpuHcclProcess::CopyCtxInfo(AicpuComContext* ctx)
577 : {
578 80 : auto otherClusterId = CLUSTER_CNT - ctx->clusterId - 1;
579 80 : auto otherCluster = &g_comContext[otherClusterId];
580 80 : *otherCluster = *ctx;
581 80 : otherCluster->clusterId = otherClusterId;
582 80 : HCCL_DEBUG("curClusterId = %d, otherClusterId = %d, copy finished", ctx->clusterId, otherCluster->clusterId);
583 80 : }
584 :
585 1 : void AicpuHcclProcess::CopyCtxForBackGroundDfx(const AicpuComContext* ctx)
586 : {
587 1 : auto otherClusterId = CLUSTER_CNT - ctx->clusterId - 1;
588 1 : AicpuComContext* otherCluster = &g_comContext[otherClusterId];
589 1 : otherCluster->workSpaceAddr = ctx->workSpaceAddr;
590 1 : otherCluster->notifyOff = ctx->notifyOff;
591 1 : otherCluster->notifyBeginCnt = ctx->notifyBeginCnt;
592 1 : otherCluster->totalTurnCnt = ctx->totalTurnCnt;
593 1 : AicpuSqeContext::SaveVariable();
594 1 : }
595 :
596 4 : HcclResult AicpuHcclProcess::WaitAsyncFlag(
597 : hccl::Transport::Buffer* localFlagBufforCheck, const uint32_t flagValue, uint64_t timeOut)
598 : {
599 : // 轮询等待flag
600 4 : if (flagValue < FLAG_OFFSET) {
601 1 : HCCL_ERROR("[AicpuHcclProcess][WaitAsyncFlag] flagValue must > 0,now flagValue is [%u]", flagValue);
602 1 : return HCCL_E_PARA;
603 : }
604 3 : uint32_t index = flagValue - FLAG_OFFSET;
605 3 : bool isTimeout = true;
606 3 : u64 startTime = GetCurCpuTimestamp();
607 3 : CHK_PTR_NULL(localFlagBufforCheck);
608 2 : CHK_PTR_NULL(localFlagBufforCheck[index].addr);
609 2 : uint32_t* waitPtr = const_cast<uint32_t*>(static_cast<const uint32_t*>(localFlagBufforCheck[index].addr));
610 2 : while ((GetCurCpuTimestamp() - startTime) < static_cast<unsigned long long>(NSEC_PER_SEC * timeOut)) {
611 1 : if (*waitPtr == flagValue) {
612 1 : isTimeout = false;
613 1 : *waitPtr = 0;
614 1 : break; // 当前位置的值等于flag时，说明对端写过来了，可以退出
615 : }
616 : }
617 2 : CHK_PRT_RET(
618 : isTimeout,
619 : HCCL_ERROR(
620 : "[AicpuHcclProcess][WaitAsyncFlag]Kernel Run TimeOut %llus, now Operation is CheckFlag, "
621 : "localFlagBufforCheck0.addr is [%p], localFlagBufforCheck0.value is [%u],"
622 : "localFlagBufforCheck1.addr is [%p], localFlagBufforCheck1.value is [%u],"
623 : "localFlagBufforCheck2.addr is [%p], localFlagBufforCheck2.value is [%u]",
624 : timeOut, localFlagBufforCheck[0].addr, *(const uint32_t*)(localFlagBufforCheck[0].addr),
625 : localFlagBufforCheck[1].addr, *(const uint32_t*)(localFlagBufforCheck[1].addr),
626 : localFlagBufforCheck[2].addr, *(const uint32_t*)(localFlagBufforCheck[2].addr)),
627 : HCCL_E_INTERNAL);
628 1 : return HCCL_SUCCESS;
629 : }
630 :
631 0 : HcclResult AicpuHcclProcess::InitAsyncFlag(
632 : const uint32_t* lFlagAddr, const uint32_t* rFlagAddr, hccl::Transport::Buffer* localFlagBufforCheck,
633 : hccl::Transport::Buffer* localFlagBufforWrite, hccl::Transport::Buffer* remoteFlagBuf)
634 : {
635 0 : CHK_PTR_NULL(lFlagAddr);
636 0 : CHK_PTR_NULL(rFlagAddr);
637 0 : CHK_PTR_NULL(remoteFlagBuf);
638 0 : for (uint32_t i = 0; i < POST_SEND_FLAG_COUNT; i++) {
639 0 : localFlagBufforCheck[i].addr = lFlagAddr + i * FLAG_INTERVAL; // 接收对端给本端写的flag
640 0 : localFlagBufforCheck[i].size = sizeof(uint32_t);
641 0 : localFlagBufforWrite[i].addr = lFlagAddr + i * FLAG_INTERVAL + FLAG_OFFSET; // 存放本端给对端写的flag
642 0 : localFlagBufforWrite[i].size = sizeof(uint32_t);
643 0 : remoteFlagBuf[i].addr = rFlagAddr + i * FLAG_INTERVAL;
644 0 : remoteFlagBuf[i].size = sizeof(uint32_t);
645 0 : uint32_t* waitPtr = const_cast<uint32_t*>(static_cast<const uint32_t*>(localFlagBufforWrite[i].addr));
646 0 : *waitPtr = i + FLAG_OFFSET; // 置flag
647 : }
648 0 : return HCCL_SUCCESS;
649 : }
650 :
651 0 : HcclResult AicpuHcclProcess::AicpuIndOpThreadInit(ThreadMgrAicpuParam* param)
652 : {
653 0 : std::string group = param->hcomId;
654 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
655 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
656 0 : HcclResult ret = hcclCommAicpu->InitThreads(param);
657 0 : CHK_PRT_RET(
658 : ret != HCCL_SUCCESS,
659 : HCCL_ERROR(
660 : "[AicpuHcclProcess][AicpuIndOpThreadInit]errNo[0x%016llx] Failed to init threads group[%s]",
661 : HCCL_ERROR_CODE(ret), group.c_str()),
662 : ret);
663 0 : AicpuReleaseCommbyGroup(group);
664 0 : return HCCL_SUCCESS;
665 0 : }
666 :
667 0 : HcclResult AicpuHcclProcess::AicpuIndOpChannelInit(HcclIndOpChannelRemoteResV3* commParam)
668 : {
669 0 : std::string group = commParam->hcomId;
670 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
671 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
672 0 : HcclResult ret = hcclCommAicpu->AllocChannelResource(commParam);
673 0 : CHK_PRT_RET(
674 : ret != HCCL_SUCCESS,
675 : HCCL_ERROR(
676 : "[AicpuHcclProcess][AicpuIndOpChannelInit]errNo[0x%016llx] Failed to init channels group[%s]",
677 : HCCL_ERROR_CODE(ret), group.c_str()),
678 : ret);
679 0 : AicpuReleaseCommbyGroup(group);
680 0 : return HCCL_SUCCESS;
681 0 : }
682 :
683 1 : HcclResult AicpuHcclProcess::AicpuIndOpNotifyInit(NotifyMgrAicpuParam* param)
684 : {
685 1 : CHK_PTR_NULL(param);
686 0 : std::string group = param->hcomId;
687 0 : hccl::HcclCommAicpu* hcclCommAicpu = AicpuHcclProcess::AicpuGetCommbyGroup(group);
688 0 : CHK_PRT_RET(!hcclCommAicpu, HCCL_ERROR("%s hcclCommAicpu is null, group[%s]", __func__, group.c_str()), HCCL_E_PTR);
689 :
690 0 : HcclResult ret = HCCL_E_INTERNAL;
691 0 : if (param->freeFlag) {
692 0 : ret = hcclCommAicpu->NotifyFree(param);
693 0 : CHK_PRT_RET(
694 : ret != HCCL_SUCCESS,
695 : HCCL_ERROR(
696 : "[AicpuHcclProcess][%s]errNo[0x%016llx] Failed to free notifys group[%s]", __func__,
697 : HCCL_ERROR_CODE(ret), group.c_str()),
698 : ret);
699 : } else {
700 0 : ret = hcclCommAicpu->NotifyAlloc(param);
701 0 : CHK_PRT_RET(
702 : ret != HCCL_SUCCESS,
703 : HCCL_ERROR(
704 : "[AicpuHcclProcess][%s]errNo[0x%016llx] Failed to alloc notifys group[%s]", __func__,
705 : HCCL_ERROR_CODE(ret), group.c_str()),
706 : ret);
707 : }
708 :
709 0 : HCCL_INFO(
710 : "[AicpuHcclProcess][%s] comm identifier[%s], notify op[%u] success, num[%u]", __func__, group.c_str(),
711 : param->freeFlag, param->notifyNum);
712 0 : AicpuReleaseCommbyGroup(group);
713 0 : return HCCL_SUCCESS;
714 0 : }
|
