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